# 機能設計書 19-並列処理

## 概要

本ドキュメントは、Nushellにおける並列処理機能の設計について記述する。par-eachコマンドを通じて、リストの各要素に対してクロージャを並列実行する機能を提供する。

### 本機能の処理概要

並列処理機能は、リストやテーブルの各要素に対してクロージャを並列に実行し、処理時間を短縮する。Rayonライブラリを使用してスレッドプールベースの並列処理を実現する。

**業務上の目的・背景**：大量のデータ処理やI/O待ちを伴う処理において、並列実行による高速化は重要な要件である。ファイルの一括処理、ネットワークリクエストの並列実行、計算負荷の高い処理の分散など、パフォーマンス向上の多様なユースケースに対応する。

**機能の利用シーン**：
- `ls | par-each {|f| open $f.name | wc -l}` でファイルを並列読み込み
- `urls | par-each {|u| http get $u}` でHTTPリクエストを並列実行
- `[1 2 3] | par-each --keep-order {|x| $x * 2}` で順序保持並列処理
- `data | par-each -t 4 {|d| expensive_calc $d}` で4スレッドで並列計算

**主要な処理内容**：
1. スレッドプールの構築
2. 入力データを並列イテレータに変換
3. 各要素に対してクロージャを並列実行
4. 結果を収集（--keep-orderで順序保持）

**関連システム・外部連携**：Rayonライブラリ（並列処理）、nu-engine（クロージャ評価）と連携。

**権限による制御**：特になし。純粋なデータ操作機能。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 1 | メインREPL | 主画面 | 並列処理コマンドの入力と結果表示 |

## 機能種別

データ操作（並列処理）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| closure | Closure | Yes | 各要素に適用するクロージャ | 有効なクロージャ |
| --threads, -t | Int | No | 使用スレッド数 | 正の整数（0=デフォルト） |
| --keep-order, -k | Switch | No | 入力順序を保持 | - |

### 入力データソース

- パイプライン入力: List、Table、Range、ByteStream
- ListStreamもサポート（par_bridge使用）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| result_list | List | クロージャ適用結果のリスト |

### 出力先

- パイプライン出力: `PipelineData`として後続コマンドへ渡す
- Nothing値は自動的にフィルタリング

## 処理フロー

### 処理シーケンス

```
1. コマンド引数の解析
   └─ クロージャ、--threads、--keep-orderを取得

2. スレッドプール構築
   └─ ThreadPoolBuilder.num_threads()でプール作成

3. 入力データの並列処理
   └─ List: into_par_iter().enumerate()
   └─ Range: into_range_iter().enumerate().par_bridge()
   └─ ListStream: into_iter().enumerate().par_bridge()

4. クロージャの並列実行
   └─ parallel_closure_map()でmap処理
   └─ ClosureEval.run_with_value()で各要素処理

5. 結果の収集とフィルタリング
   └─ --keep-order: par_sort_unstable_by_key()
   └─ Nothing値を除外

6. 結果を返却
```

### フローチャート

```mermaid
flowchart TD
    A[開始] --> B[引数解析]
    B --> C{--threads指定?}
    C -->|Yes| D[指定数でプール作成]
    C -->|No| E[デフォルトプール作成]

    D --> F[create_pool]
    E --> F

    F --> G{入力型}
    G -->|List| H[into_par_iter]
    G -->|Range| I[par_bridge]
    G -->|ListStream| J[par_bridge]
    G -->|ByteStream| K[chunks.par_bridge]
    G -->|単一Value| L[ClosureEvalOnce]

    H --> M[enumerate]
    I --> M
    J --> M
    K --> M

    M --> N[parallel_closure_map]
    N --> O[各要素にclosure適用]
    O --> P[結果収集]

    P --> Q{--keep-order?}
    Q -->|Yes| R[par_sort_unstable_by_key]
    Q -->|No| S[そのまま]

    R --> T[Nothing除外]
    S --> T
    L --> T

    T --> U[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | Nothing除外 | クロージャ結果がNothingの場合は出力から除外 | 常に適用 |
| BR-02 | エラー伝播 | クロージャ内エラーはValue::Errorとして結果に含む | 常に適用 |
| BR-03 | 順序不定 | --keep-orderなしの場合、出力順序は不定 | デフォルト |
| BR-04 | スレッド数制御 | --threads 0はRayonのデフォルト（通常は論理CPU数） | threads=0時 |

### 計算ロジック

並列処理フロー:
```
pool.install(|| {
    input.into_par_iter()
         .enumerate()
         .map_init(
             || ClosureEval::new(...),
             |closure, (idx, val)| (idx, closure.run(val))
         )
         .collect()
})
```

順序保持:
- enumerateでインデックスを付与
- --keep-orderでpar_sort_unstable_by_keyでソート
- 同一インデックスはないのでunstableソートで十分

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

該当なし（インメモリ処理のみ）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| GenericError | スレッドプールエラー | プール作成失敗 | システムリソースを確認 |

### リトライ仕様

該当なし（同期処理のため）

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

該当なし（状態変更を伴わない）

## パフォーマンス要件

- Rayonスレッドプール: ワークスティーリングで負荷分散
- --keep-order: 並列ソートで効率的に順序復元
- map_init: スレッドごとにClosureEvalを初期化（オーバーヘッド削減）

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

- クロージャ実行: エンジンのサンドボックス内で実行
- スレッド数制限: --threadsで過剰なリソース使用を防止可能

## 備考

- eachと同様のインターフェースだが順序保証なし（--keep-orderなしの場合）
- I/Oバウンドな処理で効果が高い
- 計算量の少ない処理ではオーバーヘッドが上回る可能性

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | par_each.rs | `crates/nu-command/src/filters/par_each.rs` | use文(1-5行目)でRayon依存を確認 |

**読解のコツ**: `rayon::prelude::*`のインポートでParallelIteratorトレイトが使用可能。`par_bridge()`は通常のIteratorを並列化。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | par_each.rs | `crates/nu-command/src/filters/par_each.rs` | run関数(90-192行目)で全体フローを理解 |

**主要処理フロー**:
1. **113-117行目**: closure、threads、keep_orderを取得
2. **97-111行目**: `create_pool()`でスレッドプール作成
3. **125-133行目**: `apply_order()`で順序復元関数定義
4. **135-188行目**: 入力型に応じた並列処理
5. **189行目**: `filter(!v.is_nothing())`でNothing除外

#### Step 3: 並列処理ロジックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | par_each.rs | `crates/nu-command/src/filters/par_each.rs` | parallel_closure_map関数(194-215行目) |

**主要処理フロー**:
- **200行目**: `map_init()`でスレッドごとに初期化
- **201行目**: `ClosureEval::new()`で各スレッドにクロージャ評価器
- **202-210行目**: 各要素にクロージャを適用
- **207-209行目**: エラー時はchain_error_with_inputでラップ

#### Step 4: 入力型ごとの処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | par_each.rs | `crates/nu-command/src/filters/par_each.rs` | Listの処理(140-145行目) |
| 4-2 | par_each.rs | `crates/nu-command/src/filters/par_each.rs` | Rangeの処理(146-154行目) |
| 4-3 | par_each.rs | `crates/nu-command/src/filters/par_each.rs` | ListStreamの処理(162-169行目) |

**主要処理フロー**:
- **141行目**: `vals.into_par_iter()`で並列イテレータ化
- **147-150行目**: Rangeは`par_bridge()`で並列化
- **164行目**: ListStreamも`par_bridge()`で並列化

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

```
ParEach::run()
    │
    ├─ create_pool(num_threads)
    │      └─ ThreadPoolBuilder::new().num_threads().build()
    │
    ├─ apply_order定義
    │      └─ (keep_order) par_sort_unstable_by_key
    │
    └─ pool.install(|| {...})
           │
           ├─ Value::List:
           │      └─ vals.into_par_iter()
           │             └─ enumerate()
           │             └─ parallel_closure_map()
           │                    └─ map_init(ClosureEval::new, run_with_value)
           │
           ├─ Value::Range:
           │      └─ into_range_iter()
           │             └─ enumerate().par_bridge()
           │             └─ parallel_closure_map()
           │
           ├─ ListStream:
           │      └─ into_iter()
           │             └─ enumerate().par_bridge()
           │             └─ parallel_closure_map()
           │
           └─ ByteStream:
                  └─ chunks()
                         └─ enumerate().par_bridge()
                         └─ parallel_closure_map()
```

### データフロー図

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

                       ┌─────────────────────────────────┐
PipelineData ─────────▶│       スレッドプール構築          │
     │                 │  create_pool(--threads)         │
     │                 └─────────────────────────────────┘
     │                                  │
     │                                  ▼
     │                 ┌─────────────────────────────────┐
     │                 │       並列イテレータ変換         │
     │                 │  List: into_par_iter()          │
     │                 │  Other: par_bridge()            │
     │                 └─────────────────────────────────┘
     │                                  │
     │                                  ▼
     │                 ┌─────────────────────────────────┐
     │                 │       並列クロージャ実行         │
     │                 │  ┌───────┬───────┬───────┐      │
     │                 │  │Thread │Thread │Thread │      │
     │                 │  │  1    │  2    │  ...  │      │
     │                 │  │closure│closure│closure│      │
     │                 │  │ eval  │ eval  │ eval  │      │
     │                 │  └───────┴───────┴───────┘      │
     │                 └─────────────────────────────────┘
     │                                  │
     │                                  ▼
     │                 ┌─────────────────────────────────┐
     │                 │        結果収集・整列            │
     │                 │  (--keep-order) sort by index   │
     │                 │  filter(!is_nothing)            │
     │                 └─────────────────────────────────┘
     │                                  │
     └──────────────────────────────────▶ List (results)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| par_each.rs | `crates/nu-command/src/filters/par_each.rs` | ソース | par-eachコマンド実装 |
| utils.rs | `crates/nu-command/src/filters/utils.rs` | ソース | chain_error_with_inputヘルパー |
| mod.rs | `crates/nu-command/src/filters/mod.rs` | ソース | フィルターモジュールエクスポート |
