# 機能設計書 16-バリア実行モード

## 概要

本ドキュメントは、Apache Sparkのバリア実行モードについて、分散機械学習で必要なバリア同期をサポートする実行モードの設計仕様を記述した機能設計書である。

### 本機能の処理概要

分散機械学習フレームワーク（MPI的な同期パターン）で必要なバリア同期をサポートし、全タスクが同期ポイントに到達するまで待機する実行モードを提供する。BarrierCoordinator（ドライバー側）とBarrierTaskContext（タスク側）の協調により、グローバルバリア同期を実現する。

**業務上の目的・背景**：分散機械学習ライブラリ（Horovod、PyTorch Distributed等）では、全ワーカーが同じフェーズで同期する「バリア同期」が必要である。従来のMapReduce型のSpark処理モデルでは、タスクは独立して実行されるが、バリア実行モードにより全タスクが同期ポイントに到達してから次のフェーズに進む仕組みを提供する。

**機能の利用シーン**：(1) 分散機械学習の勾配同期、(2) BSP（Bulk Synchronous Parallel）パターンの分散計算、(3) 全タスク間でのメッセージ交換（allGather操作）。

**主要な処理内容**：
1. BarrierCoordinatorのRpcEndpointとしての登録とバリアリクエスト管理
2. BarrierTaskContextからのbarrier()呼び出しによる同期リクエスト送信
3. 全タスクのリクエスト集約とタイムアウト管理
4. バリアエポックによる複数回のbarrier()呼び出し管理
5. allGather()による全タスク間メッセージ交換
6. ステージ完了時のバリアステート清掃

**関連システム・外部連携**：RPC通信フレームワーク（BarrierCoordinatorはRpcEndpointとして動作）

**権限による制御**：特になし

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | バリア実行モードは画面と直接関連しない基盤機能 |

## 機能種別

分散同期 / 実行制御

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| spark.barrier.sync.timeout | Long | No | バリア同期のタイムアウト(秒) | デフォルト設定に依存 |
| numTasks | Int | Yes | バリアステージのタスク数 | RequestToSyncのnumTasksと一致すること |
| stageId | Int | Yes | ステージID | - |
| stageAttemptId | Int | Yes | ステージ試行ID | - |
| taskAttemptId | Long | Yes | タスク試行ID | - |
| barrierEpoch | Int | Yes | バリアエポック（barrier()呼び出し回数） | 現在のCoordinator側エポックと一致 |
| message | String | No | タスク間で交換するメッセージ | allGather使用時 |

### 入力データソース

BarrierTaskContext.barrier()またはBarrierTaskContext.allGather()からのRPCリクエスト。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| messages | Array[String] | allGather()で各タスクから集約されたメッセージ配列 |
| 同期完了通知 | Unit | barrier()の同期完了（全タスクへの応答） |

### 出力先

RPCレスポンスとして各BarrierTaskContextに返却。

## 処理フロー

### 処理シーケンス

```
1. BarrierCoordinatorの初期化
   └─ SparkContextのRpcEnvにRpcEndpointとして登録
2. バリアステージの実行開始
   └─ 各タスクがBarrierTaskContextを通じてbarrier()/allGather()を呼び出し
3. リクエストの集約
   └─ BarrierCoordinator.receiveAndReply()でRequestToSyncを受信し、ContextBarrierStateに蓄積
4. タイムアウト管理
   └─ 最初のリクエスト受信時にタイマーを開始
5. 全タスク到達判定
   └─ requesters.size == numTasks で全タスクからのリクエスト到達を確認
6. 応答と次エポック準備
   └─ 全タスクに応答を返し、barrierEpochをインクリメント
7. ステージ完了時の清掃
   └─ SparkListenerStageCompletedイベントでContextBarrierStateをクリア
```

### フローチャート

```mermaid
flowchart TD
    A[BarrierTaskContext.barrier] --> B[RequestToSync RPC送信]
    B --> C[BarrierCoordinator.receiveAndReply]
    C --> D[ContextBarrierState.handleRequest]
    D --> E{最初のリクエスト?}
    E -->|Yes| F[タイマー開始]
    E -->|No| G[リクエスト追加]
    F --> G
    G --> H{全タスク到達?}
    H -->|Yes| I[全タスクに応答返送]
    I --> J[barrierEpoch++]
    J --> K[タイマーキャンセル]
    H -->|No| L[待機]
    L --> M{タイムアウト?}
    M -->|Yes| N[SparkException送信]
    N --> O[ステート清掃]
    M -->|No| L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-16-01 | 単一アクティブバリア | 各バリアステージ試行では同時に1つのbarrier()呼び出しのみアクティブ | 全バリアステージ |
| BR-16-02 | エポック一致 | タスクのbarrierEpochがCoordinator側と一致しない場合は失敗 | barrier()呼び出し時 |
| BR-16-03 | 全タスク同一メソッド | 同一バリア同期フェーズ内では全タスクが同じメソッド（barrier/allGather）を呼ぶこと | barrier()呼び出し時 |
| BR-16-04 | タイムアウト失敗 | タイムアウト時間内に全タスクからリクエストが届かない場合、全リクエストを失敗 | タイムアウト発生時 |
| BR-16-05 | ステージ完了清掃 | ステージ完了時にContextBarrierStateを清掃 | SparkListenerStageCompleted時 |

### 計算ロジック

該当なし

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

### テーブル別操作詳細

該当なし

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | SparkException | バリア同期のタイムアウト | 全リクエスターにsendFailureで通知、ステートクリア |
| - | SparkException | エポック不一致 | 該当タスクにsendFailureで通知 |
| - | SparkException | 同一フェーズで異なるメソッド使用 | 全リクエスターにsendFailureで通知、ステートクリア |
| - | IllegalArgumentException | numTasks不一致 | require失敗で例外スロー |
| - | TaskKilledException | タスクがkillされた場合 | BarrierTaskContext側で検出しRPCをアボート |

### リトライ仕様

バリア同期失敗時はステージ全体のリトライ（別のstageAttemptIdで再実行）となる。個別タスクのリトライはバリアモードでは行わない。

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

該当なし。ContextBarrierStateはsynchronizedブロックで排他制御される。

## パフォーマンス要件

バリア同期の待ち時間は全タスクの到達時間に依存する。タスク間の処理時間差が大きい場合、待ち時間が増加する。1分ごとに待機状況がログ出力される。

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

特になし。RPC通信のセキュリティ設定に依存。

## 備考

- BarrierTaskContextはExperimental APIとして@Experimental アノテーション付き
- RPCタイムアウトは365日に設定され、実質的にCoordinator側のタイムアウトで制御
- RequestMethod列挙型にBARRIERとALL_GATHERの2種がある

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | BarrierCoordinator.scala | `core/src/main/scala/org/apache/spark/BarrierCoordinator.scala` | ContextBarrierId、RequestToSync、RequestMethodの定義 |

**読解のコツ**: ContextBarrierIdは(stageId, stageAttemptId)のタプルで一意識別。ContextBarrierStateはバリア同期の状態マシンとして機能する。

#### Step 2: エントリーポイントを理解する（タスク側）

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | BarrierTaskContext.scala | `core/src/main/scala/org/apache/spark/BarrierTaskContext.scala` | barrier(), allGather()メソッドの実装、runBarrier()の処理フロー |

**主要処理フロー**:
1. **52-55行目**: barrierCoordinator RpcEndpointRefの取得
2. **59行目**: barrierEpochのローカル管理
3. **71-93行目**: runBarrier()でRequestToSyncをRPC送信し、応答を待機
4. **88-93行目**: askAbortable()で1秒ごとにTaskKill確認しながら待機

#### Step 3: コーディネーター側を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | BarrierCoordinator.scala | `core/src/main/scala/org/apache/spark/BarrierCoordinator.scala` | receiveAndReply(), ContextBarrierState.handleRequest()の処理フロー |

**主要処理フロー**:
- **50-53行目**: ThreadSafeRpcEndpointとして初期化、タイムアウト設定
- **101-103行目**: ContextBarrierStateのコンストラクタ（barrierId, numTasks）
- **147-200行目**: handleRequest()のメインロジック（リクエスト検証→蓄積→全タスク到達判定→応答）
- **187-198行目**: 全タスク到達時にreply()で応答、エポックインクリメント、タイマーキャンセル
- **220-229行目**: receiveAndReply()でRequestToSyncを受信しstatesから状態取得

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

```
[タスク側]                          [ドライバー側]
BarrierTaskContext
    │
    ├─ barrier()
    │    └─ runBarrier(message, BARRIER)
    │         └─ barrierCoordinator.askAbortable(RequestToSync)
    │              │
    ├─ allGather(message)                  BarrierCoordinator (RpcEndpoint)
    │    └─ runBarrier(message, ALL_GATHER)      │
    │         └─ barrierCoordinator.askAbortable  ├─ receiveAndReply(RequestToSync)
    │              │                               │    └─ states.get(barrierId)
    │              │                               │         └─ ContextBarrierState
    │              └──── RPC ──────────────────────┘              │
    │                                                              ├─ handleRequest()
    │                                                              │    ├─ requestMethod検証
    │                                                              │    ├─ epoch検証
    │                                                              │    ├─ タイマー管理
    │                                                              │    └─ 全タスク到達→reply
    │                                                              │
    └─ 応答受信 ◀─────── RPC reply ─────────────────────────────┘

SparkListener
    └─ onStageCompleted → cleanupBarrierStage()
```

### データフロー図

```
[タスク側]                    [ドライバー側]                  [結果]

barrier()呼び出し ───▶ RequestToSync           ───▶ ContextBarrierState
(全タスク)              (stageId, epoch等)            ├─ requesters蓄積
                                                       ├─ message蓄積
                                                       └─ 全タスク到達判定

                        ContextBarrierState     ───▶ reply(messages)
                        (全タスク到達時)                → 各タスクへ応答
                                                       → epoch++
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| BarrierCoordinator.scala | `core/src/main/scala/org/apache/spark/BarrierCoordinator.scala` | ソース | バリア同期コーディネーター（ドライバー側） |
| BarrierTaskContext.scala | `core/src/main/scala/org/apache/spark/BarrierTaskContext.scala` | ソース | バリアタスクコンテキスト（タスク側） |
| TaskContext.scala | `core/src/main/scala/org/apache/spark/TaskContext.scala` | ソース | 基底タスクコンテキスト |
| RpcEndpoint.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEndpoint.scala` | ソース | ThreadSafeRpcEndpointの定義 |
