# 機能設計書 92-Network Communication Infrastructure

## 概要

本ドキュメントは、Apache Sparkのネットワーク通信基盤（Network Communication Infrastructure）の設計について記述する。Sparkの全コンポーネント間で使用される低レベルネットワーク通信フレームワークを提供する。

### 本機能の処理概要

**業務上の目的・背景**：Sparkはドライバー・エグゼキューター間でRPCメッセージやシャッフルデータを効率的に転送する必要がある。本機能はNettyベースの高性能なネットワーク通信レイヤーを提供し、接続管理、メッセージのシリアライズ/デシリアライズ、チャンクフェッチ、ストリーミング、SSL/TLSサポートなどを統一的なAPIで扱えるようにする。

**機能の利用シーン**：RPCメッセージの送受信（ジョブ投入、タスク結果の返送、ハートビート等）、シャッフルデータのフェッチ、ブロック転送、外部シャッフルサービスとの通信に利用される。

**主要な処理内容**：
1. TransportContextを通じたサーバー/クライアントの生成とNettyパイプラインの初期化
2. TransportServerによるNettyサーバーのバインドと接続受付
3. TransportClientFactoryによる接続プール管理とクライアント生成
4. TransportChannelHandlerによるリクエスト/レスポンスの双方向ディスパッチ
5. RpcHandlerによるアプリケーション固有のRPCメッセージ処理

**関連システム・外部連携**：Sparkの全コンポーネント（RPC、シャッフル、ブロック転送）から利用される。Nettyフレームワーク（EventLoopGroup、ChannelPipeline）に依存。

**権限による制御**：SSL/TLSによる暗号化通信をサポート。SASL認証のためのTransportServerBootstrap/TransportClientBootstrapを提供。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 画面機能マッピングに本機能に対応する画面定義なし |

## 機能種別

基盤 / ネットワーク通信

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| hostToBind | String | No | サーバーのバインドホスト（nullで全インターフェース） | IPアドレスまたはホスト名 |
| portToBind | int | Yes | サーバーのバインドポート（0で自動割り当て） | 0-65535 |
| conf | TransportConf | Yes | トランスポート設定 | - |
| appRpcHandler | RpcHandler | Yes | アプリケーションRPCハンドラー | - |
| bootstraps | List | No | サーバー/クライアントブートストラップ | - |

### 入力データソース

Nettyチャネルを通じたバイナリメッセージ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| TransportClient | TransportClient | リモートホストへの接続クライアント |
| TransportServer | TransportServer | リスニングサーバーインスタンス |
| RpcResponseCallback | ByteBuffer | RPCレスポンスデータ |

### 出力先

Nettyチャネル経由でリモートホストへ送信

## 処理フロー

### 処理シーケンス

```
1. TransportContext生成: TransportConfとRpcHandlerを受け取る
2. サーバー側:
   a. TransportContext.createServer() → TransportServer生成
   b. TransportServer.init(): Netty ServerBootstrapの設定
      - IOMode決定（NIO/EPOLL）
      - EventLoopGroup生成（boss/worker）
      - ChannelOptionの設定（SO_REUSEADDR, SO_BACKLOG, SO_KEEPALIVE等）
   c. ChannelInitializerでNettyパイプラインを初期化
      - TransportServerBootstrap.doBootstrap()でカスタマイズ
      - TransportContext.initializePipeline()でハンドラー設置
3. クライアント側:
   a. TransportContext.createClientFactory() → TransportClientFactory生成
   b. TransportClientFactory.createClient(): 接続プールからクライアント取得/生成
      - 接続プールチェック → 既存クライアントがあれば再利用
      - Netty Bootstrapでリモート接続
      - TransportClientBootstrap.doBootstrap()でカスタマイズ
4. メッセージ処理:
   a. TransportChannelHandler.channelRead0(): メッセージディスパッチ
      - RequestMessage → TransportRequestHandler
      - ResponseMessage → TransportResponseHandler
   b. RpcHandler.receive(): アプリケーション固有のRPC処理
```

### フローチャート

```mermaid
flowchart TD
    A[TransportContext生成] --> B[サーバー作成]
    A --> C[クライアントファクトリ作成]
    B --> D[Netty ServerBootstrap設定]
    D --> E[バインド・リスン開始]
    C --> F{接続プールに既存あり?}
    F -->|Yes| G[既存クライアント返却]
    F -->|No| H[Netty Bootstrap接続]
    H --> I[パイプライン初期化]
    I --> J[TransportClient生成]
    E --> K[接続受付]
    K --> L[パイプライン初期化]
    L --> M[TransportChannelHandler]
    M --> N{メッセージ種別}
    N -->|Request| O[TransportRequestHandler]
    N -->|Response| P[TransportResponseHandler]
    O --> Q[RpcHandler.receive]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-92-01 | 接続プール | 同一リモートホストへの接続をプーリングして再利用 | TransportClientFactory使用時 |
| BR-92-02 | Fast Fail | 直近の接続失敗から一定時間内は即座に失敗させる | fastFailTimeWindow > 0 |
| BR-92-03 | 双方向通信 | クライアント・サーバー間のチャネルは双方向で使用可能 | 常時 |
| BR-92-04 | アイドルタイムアウト | 未処理リクエストがある状態でアイドルになるとタイムアウト | requestTimeoutMs設定 |
| BR-92-05 | バッファアロケータ共有 | 共有ByteBufアロケータの使用が設定可能 | sharedByteBufAllocators=true |

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

本機能はデータベースを直接操作しない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | 接続エラー | リモートホストに接続不可 | Fast Fail機構で連続失敗を回避 |
| IdleStateEvent | タイムアウト | アイドルタイムアウト超過 | closeIdleConnectionsがtrueの場合チャネルクローズ |
| IllegalStateException | 初期化エラー | サーバー未初期化でポート取得 | サーバーの初期化を確認 |

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

個々のRPCは要求-応答のペアで完結する。ストリーミング転送はチャンク単位で処理される。

## パフォーマンス要件

- IOMode: NIOまたはEPOLL（Linux環境ではEPOLLが推奨）
- 接続プールサイズ: numConnectionsPerPeerで設定
- PooledByteBufAllocatorによる効率的なメモリ管理
- NettyMemoryMetricsによるメモリ使用状況の監視

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

- SSL/TLS: SslHandlerによる暗号化通信
- SASL認証: TransportServerBootstrap/TransportClientBootstrapでの認証プラグイン

## 備考

`common/network-common/` パッケージ配下に実装されている。Sparkの全ネットワーク通信の基盤となるモジュール。

---

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Message.java | `common/network-common/src/main/java/org/apache/spark/network/protocol/Message.java` | メッセージプロトコルの基底インターフェース |
| 1-2 | RequestMessage.java / ResponseMessage.java | `common/network-common/src/main/java/org/apache/spark/network/protocol/` | リクエスト・レスポンスメッセージの型定義 |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TransportContext.java | `common/network-common/src/main/java/org/apache/spark/network/TransportContext.java` | サーバー/クライアント/パイプラインの生成。initializePipeline()でNettyパイプライン構築 |

**主要処理フロー**:
- TransportContext.createServer(): TransportServerの生成
- TransportContext.createClientFactory(): TransportClientFactoryの生成
- TransportContext.initializePipeline(): Nettyパイプラインへのハンドラー登録

#### Step 3: サーバー実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TransportServer.java | `common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java` | **99-156行目**: init()メソッドでNetty ServerBootstrapの設定、IOMode決定、EventLoopGroup生成、ChannelOption設定、バインド |

#### Step 4: クライアント実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | TransportClientFactory.java | `common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java` | **66-79行目**: ClientPoolデータ構造（接続プール）。createClient()で接続プール管理とNettyクライアント生成 |

#### Step 5: メッセージハンドリングを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | TransportChannelHandler.java | `common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java` | **38-54行目**: 双方向通信のアーキテクチャ説明。channelRead0()でRequest/Responseのディスパッチ |
| 5-2 | RpcHandler.java | `common/network-common/src/main/java/org/apache/spark/network/server/RpcHandler.java` | **33-55行目**: アプリケーション固有RPCの抽象ハンドラー。receive()でRPCメッセージ処理 |

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

```
TransportContext
    |
    +-- createServer(host, port, rpcHandler, bootstraps)
    |     +-- TransportServer(context, host, port, rpcHandler, bootstraps)
    |           +-- init(host, port)
    |                 +-- NettyUtils.createEventLoop() [boss, worker]
    |                 +-- ServerBootstrap.bind()
    |                 +-- ChannelInitializer.initChannel()
    |                       +-- TransportServerBootstrap.doBootstrap()
    |                       +-- TransportContext.initializePipeline()
    |
    +-- createClientFactory(bootstraps)
    |     +-- TransportClientFactory(context, bootstraps)
    |           +-- createClient(remoteHost, remotePort)
    |                 +-- connectionPool.get/put [接続プール]
    |                 +-- Bootstrap.connect()
    |                 +-- TransportClientBootstrap.doBootstrap()
    |
    +-- initializePipeline(channel, rpcHandler)
          +-- TransportChannelHandler
                +-- TransportRequestHandler
                +-- TransportResponseHandler
```

### データフロー図

```
[クライアント側]                  [ネットワーク]          [サーバー側]

TransportClient                                    TransportServer
  .sendRpc()      ──RequestMessage──>            TransportChannelHandler
  .fetchChunk()                                    .channelRead0()
                                                      |
                                                TransportRequestHandler
                                                  .handle()
                                                      |
                                                  RpcHandler
                                                    .receive()
                                                      |
TransportResponseHandler <──ResponseMessage──  RpcResponseCallback
  .handle()                                      .onSuccess()
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| TransportContext.java | `common/network-common/src/main/java/org/apache/spark/network/TransportContext.java` | ソース | サーバー/クライアント/パイプラインの生成 |
| TransportServer.java | `common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java` | ソース | Nettyサーバーのライフサイクル管理 |
| TransportClientFactory.java | `common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java` | ソース | 接続プール管理とクライアント生成 |
| TransportClient.java | `common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java` | ソース | リモートホストとの通信クライアント |
| TransportChannelHandler.java | `common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java` | ソース | メッセージディスパッチャー |
| TransportRequestHandler.java | `common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java` | ソース | リクエスト処理 |
| TransportResponseHandler.java | `common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java` | ソース | レスポンス処理 |
| RpcHandler.java | `common/network-common/src/main/java/org/apache/spark/network/server/RpcHandler.java` | ソース | アプリケーションRPCの抽象ハンドラー |
| TransportConf.java | `common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java` | ソース | トランスポート設定 |
| NettyUtils.java | `common/network-common/src/main/java/org/apache/spark/network/util/NettyUtils.java` | ソース | Nettyユーティリティ |
| Message.java | `common/network-common/src/main/java/org/apache/spark/network/protocol/Message.java` | ソース | メッセージプロトコル |
