# 機能設計書 13-RPC通信フレームワーク

## 概要

本ドキュメントは、Apache SparkのNettyベースRPC通信フレームワークについて、ドライバーとExecutor間の通信基盤の設計仕様を記述した機能設計書である。

### 本機能の処理概要

Nettyベースのリモートプロシージャコール（RPC）機構を提供し、Sparkクラスタ内のドライバーとExecutor間、およびコンポーネント間のメッセージ通信を管理する。RpcEnv、RpcEndpoint、RpcEndpointRefの3層構造により、抽象化されたメッセージパッシングを実現する。

**業務上の目的・背景**：Sparkは分散処理フレームワークであり、ドライバーとExecutor間、マスターとワーカー間など、複数のプロセス間で頻繁にメッセージのやり取りを行う必要がある。このRPC通信フレームワークは、タスクの割り当て、ハートビート、ステータス報告、シャッフルデータの転送調整など、全てのコンポーネント間通信の基盤となる。

**機能の利用シーン**：(1) ドライバーからExecutorへのタスク送信、(2) Executorからドライバーへのタスクステータス報告、(3) マスターとワーカー間のリソース管理通信、(4) BarrierCoordinatorによるバリア同期、(5) ブロックマネージャ間のブロック転送調整。

**主要な処理内容**：
1. RpcEnvの生成（NettyRpcEnvFactory経由）とバインドアドレスの確立
2. RpcEndpointの登録と名前解決によるメッセージルーティング
3. 非同期メッセージ送信（send/fire-and-forget）と同期応答（ask/request-reply）
4. リモートノードへの接続確立・切断・エラーハンドリング
5. ファイルサーバー機能（ドライバーからExecutorへのファイル配布）

**関連システム・外部連携**：Nettyライブラリ（ネットワークI/O）、common/network-common（低レベルネットワーク通信基盤）

**権限による制御**：SecurityManagerを通じた認証・暗号化設定に対応。SSL/TLSによる通信暗号化をサポート。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | RPC通信フレームワークは画面とは直接関連しない基盤機能 |

## 機能種別

基盤通信 / フレームワーク

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| name | String | Yes | RpcEnvの名前 | 空でないこと |
| bindAddress | String | Yes | バインドアドレス | 有効なホスト名/IP |
| advertiseAddress | String | Yes | 外部公開アドレス | 有効なホスト名/IP |
| port | Int | Yes | ポート番号 | 0-65535 |
| numUsableCores | Int | No | 利用可能コア数 | 0以上 |
| clientMode | Boolean | No | クライアントモードか否か | デフォルト: false |
| spark.rpc.lookupTimeout | Duration | No | RPCルックアップタイムアウト | デフォルト: 120s |

### 入力データソース

Sparkコンポーネント内部から送信されるメッセージオブジェクト（Serializable）。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| RpcEndpointRef | RpcEndpointRef | 登録されたエンドポイントへの参照 |
| RpcAddress | RpcAddress | RpcEnvのリスニングアドレス |
| 応答メッセージ | Any | ask()によるリクエスト・レスポンスの応答 |

### 出力先

RPCメッセージの送信先エンドポイント。ローカルまたはリモートのRpcEndpoint実装。

## 処理フロー

### 処理シーケンス

```
1. RpcEnvの生成
   └─ RpcEnv.create()でNettyRpcEnvFactoryを使用してRpcEnvインスタンス生成
2. RpcEndpointの登録
   └─ setupEndpoint(name, endpoint)でエンドポイントを名前付きで登録
3. RpcEndpointRefの取得
   └─ setupEndpointRef(address, name)でリモートエンドポイントへの参照を取得
4. メッセージ送信
   └─ send(): fire-and-forget方式、ask(): リクエスト・レスポンス方式
5. メッセージ受信・処理
   └─ RpcEndpoint.receive()またはreceiveAndReply()でメッセージを処理
6. 接続ライフサイクル管理
   └─ onConnected, onDisconnected, onNetworkErrorコールバック
7. シャットダウン
   └─ RpcEnv.shutdown()で非同期にRpcEnvを停止
```

### フローチャート

```mermaid
flowchart TD
    A[RpcEnv.create] --> B[NettyRpcEnvFactory.create]
    B --> C[NettyRpcEnv初期化]
    C --> D[setupEndpoint で RpcEndpoint登録]
    D --> E{メッセージ送信方式}
    E -->|send fire-and-forget| F[RpcEndpoint.receive]
    E -->|ask request-reply| G[RpcEndpoint.receiveAndReply]
    G --> H[RpcCallContext.reply で応答]
    F --> I[処理完了]
    H --> I
    I --> J{接続イベント}
    J -->|接続| K[onConnected]
    J -->|切断| L[onDisconnected]
    J -->|エラー| M[onNetworkError]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-13-01 | エンドポイントライフサイクル | constructor -> onStart -> receive* -> onStopの順序を保証 | 全RpcEndpoint |
| BR-13-02 | スレッドセーフティ | ThreadSafeRpcEndpointはメッセージの逐次処理を保証 | ThreadSafeRpcEndpoint実装時 |
| BR-13-03 | IsolatedEndpoint | IsolatedRpcEndpointは専用スレッドプールでメッセージ配信 | IsolatedRpcEndpoint実装時 |
| BR-13-04 | タイムアウト制御 | RpcTimeoutによるルックアップ・通信のタイムアウト管理 | 全RPC通信 |
| BR-13-05 | ファイルサーバーURI | ディレクトリURIに"/files"や"/jars"は使用不可 | RpcEnvFileServer |

### 計算ロジック

該当なし

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

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

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

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

該当なし

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | SparkException | receive/receiveAndReply未実装のメッセージ受信 | onError()に委任 |
| - | RpcEndpointNotFoundException | 存在しないエンドポイントへのアクセス | 例外スロー |
| - | RpcEnvStoppedException | 停止済みRpcEnvの使用 | 例外スロー |
| - | RpcTimeoutException | RPC通信のタイムアウト | タイムアウト例外として上位に通知 |

### リトライ仕様

RPCフレームワーク自体にはリトライ機構は組み込まれていない。リトライは各使用側で実装する。

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

該当なし（メッセージパッシング方式）

## パフォーマンス要件

Nettyベースの非同期I/Oにより高スループットを実現。スレッドプールサイズはnumUsableCoresに基づいて設定される。

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

- SecurityManagerを通じたネットワーク認証の有効化（spark.authenticate）
- SSL/TLS通信の暗号化設定（spark.ssl.rpc.enabled）
- RpcEnvFileServerによるファイル配布のアクセス制御

## 備考

- 以前はAkkaベースだったが、現在はNettyベースの独自実装に移行済み
- IsolatedThreadSafeRpcEndpointはthreadCount=1に固定されたIsolatedRpcEndpoint
- RpcEndpointRefのデシリアライゼーションにはRpcEnvコンテキストが必要

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | RpcEndpoint.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEndpoint.scala` | RpcEndpoint, ThreadSafeRpcEndpoint, IsolatedRpcEndpointの型階層とライフサイクル |
| 1-2 | RpcEndpointRef.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala` | send(), ask()メソッドのインターフェース |
| 1-3 | RpcAddress.scala | `core/src/main/scala/org/apache/spark/rpc/RpcAddress.scala` | ホスト:ポートのアドレス表現 |
| 1-4 | RpcCallContext.scala | `core/src/main/scala/org/apache/spark/rpc/RpcCallContext.scala` | reply(), sendFailure()のコールバックインターフェース |

**読解のコツ**: Scalaのトレイトの多重継承に注意。RpcEndpoint -> ThreadSafeRpcEndpoint、RpcEndpoint -> IsolatedRpcEndpoint -> IsolatedThreadSafeRpcEndpointの継承関係を把握すること。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | RpcEnv.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala` | RpcEnvの抽象クラス定義。create(), setupEndpoint(), setupEndpointRef()等のAPI |

**主要処理フロー**:
1. **37-59行目**: RpcEnv.create()でNettyRpcEnvFactoryを使用してインスタンス生成
2. **72-151行目**: RpcEnv抽象クラスのAPI定義（setupEndpoint, asyncSetupEndpointRefByURI等）
3. **159-222行目**: RpcEnvFileServerのファイル配布インターフェース

#### Step 3: Netty実装層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | NettyRpcEnv.scala | `core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala` | RpcEnvの具象実装。メッセージのディスパッチ、接続管理 |

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

```
SparkEnv / SparkContext
    │
    ├─ RpcEnv.create()
    │      └─ NettyRpcEnvFactory.create(config)
    │           └─ NettyRpcEnv (Nettyサーバー起動)
    │
    ├─ rpcEnv.setupEndpoint(name, endpoint)
    │      └─ Dispatcher に RpcEndpoint を登録
    │           └─ Inbox 作成 → onStart() 呼び出し
    │
    ├─ rpcEnv.setupEndpointRef(address, name)
    │      └─ リモートエンドポイントへの参照を取得
    │
    ├─ endpointRef.send(message) ─ fire-and-forget
    │      └─ Outbox → TransportClient → ネットワーク送信
    │
    ├─ endpointRef.ask(message) ─ request-reply
    │      └─ Outbox → TransportClient → 応答待ち → Future完了
    │
    └─ rpcEnv.shutdown()
         └─ TransportServer停止 → Dispatcher停止
```

### データフロー図

```
[送信側]                        [通信層]                    [受信側]

RpcEndpointRef.send() ───▶ Outbox              ───▶ Inbox ───▶ RpcEndpoint.receive()
                            │                        │
RpcEndpointRef.ask()  ───▶ TransportClient    ───▶ TransportServer
                            │                        │
                            └─ Netty Channel ────────┘
                                (TCP/SSL)

RpcEndpoint.receiveAndReply() ◀── RpcCallContext.reply() ◀── 応答メッセージ
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| RpcEnv.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala` | ソース | RpcEnv抽象定義・ファクトリ |
| RpcEndpoint.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEndpoint.scala` | ソース | RpcEndpointトレイト定義 |
| RpcEndpointRef.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala` | ソース | エンドポイント参照の抽象定義 |
| RpcCallContext.scala | `core/src/main/scala/org/apache/spark/rpc/RpcCallContext.scala` | ソース | RPC応答コンテキスト |
| RpcAddress.scala | `core/src/main/scala/org/apache/spark/rpc/RpcAddress.scala` | ソース | RPCアドレス表現 |
| RpcTimeout.scala | `core/src/main/scala/org/apache/spark/rpc/RpcTimeout.scala` | ソース | RPCタイムアウト管理 |
| NettyRpcEnv.scala | `core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala` | ソース | Nettyベース実装 |
| RpcEndpointAddress.scala | `core/src/main/scala/org/apache/spark/rpc/RpcEndpointAddress.scala` | ソース | エンドポイントアドレス |
