# 機能設計書 72-分散コーディネータ

## 概要

本ドキュメントは、TensorFlowにおける分散コーディネータ（Distribute Coordinator）機能の設計について記述する。分散コーディネータは、マルチワーカー環境でのワーカー調整・タスク管理・サーバー起動を行うコンポーネントである。

### 本機能の処理概要

**業務上の目的・背景**：分散TensorFlowでは、複数のワーカー、パラメータサーバー、エバリュエータが協調して学習を行う。分散コーディネータは、TF_CONFIG環境変数やClusterSpecからクラスタ構成を読み取り、各タスクの役割に応じた適切な実行環境をセットアップする。これにより、ユーザは複雑な分散設定を意識せずに分散学習を行える。

**機能の利用シーン**：MultiWorkerMirroredStrategyやParameterServerStrategyを使用した分散学習時に内部的に呼び出される。STANDALONE_CLIENTモードではクライアントがリモートサーバーに接続して学習を制御し、INDEPENDENT_WORKERモードでは各ワーカーが独立して動作する。

**主要な処理内容**：
1. TF_CONFIG環境変数またはClusterSpecからクラスタ構成を解決する
2. コーディネーションモード（STANDALONE_CLIENT / INDEPENDENT_WORKER）に基づく実行制御
3. ワーカースレッドの生成と管理（between-graphレプリケーション時）
4. TensorFlowサーバーの起動と管理
5. エバリュエータタスクの分離実行
6. ワーカー間の同期バリア機能の提供

**関連システム・外部連携**：TF_CONFIG環境変数、gRPC通信、TensorFlowサーバー（server_lib.Server）

**権限による制御**：タスクタイプ（chief、worker、ps、evaluator、client）によって実行される処理が異なる。chiefノードは初期化操作やチェックポイント保存を担当する。

## 関連画面

本機能はバックエンドの分散制御コンポーネントであるため、直接的な関連画面は存在しない。

## 機能種別

制御処理 / プロセス管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| worker_fn | Callable | Yes | ワーカーが実行する関数 | strategy引数を受け取ること |
| strategy | DistributionStrategy | Yes | 分散ストラテジーオブジェクト | configureメソッドを持つこと |
| eval_fn | Callable | No | エバリュエータが実行する関数 | 省略時はworker_fnが使用される |
| eval_strategy | DistributionStrategy | No | エバリュエータ用ストラテジー | 省略可 |
| mode | CoordinatorMode | No | コーディネーションモード | STANDALONE_CLIENT / INDEPENDENT_WORKER |
| cluster_spec | dict/ClusterDef/ClusterSpec | No | クラスタ構成 | 省略時はTF_CONFIGから読み取り |
| task_type | string | No | 現在のタスクタイプ | chief/worker/ps/evaluator/client |
| task_id | int | No | 現在のタスクID | 0以上の整数 |
| session_config | ConfigProto | No | セッション設定 | 省略時はデフォルト値 |
| rpc_layer | string | No | RPCプロトコル | デフォルト"grpc" |

### 入力データソース

- TF_CONFIG環境変数（JSON形式のクラスタ構成）
- プログラム引数として直接指定されたClusterSpec

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| worker_fn戻り値 | Any | in-graphレプリケーションまたはINDEPENDENT_WORKERモード時のworker_fnの戻り値 |
| None | NoneType | between-graphレプリケーション時 |
| Server | tf.distribute.Server | run_standard_tensorflow_server時のサーバーオブジェクト |

### 出力先

呼び出し元への戻り値

## 処理フロー

### 処理シーケンス

```
1. クラスタ構成の解決
   └─ TF_CONFIG環境変数またはcluster_spec引数からクラスタ情報を取得
   └─ cluster_resolverからの解決（strategy.extended._cluster_resolver使用時）
2. セッション設定の初期化
   └─ ConfigProtoの生成とallow_soft_placement設定
3. モード判定による分岐
   ├─ ローカルモード（cluster_specなし）
   │   └─ worker_fnを直接実行、eval_fnも直接実行
   ├─ STANDALONE_CLIENTモード
   │   ├─ クライアントタスク
   │   │   ├─ between-graph: ワーカースレッドを生成してworker_fnを並列実行
   │   │   └─ in-graph: 単一スレッドでworker_fnを実行
   │   └─ 非クライアントタスク
   │       └─ 標準TFサーバーを起動してjoin()で待機
   └─ INDEPENDENT_WORKERモード
       ├─ 全ノードで標準TFサーバーを起動
       ├─ chief/worker: worker_fnを実行
       ├─ evaluator: eval_fnを実行
       └─ ps: サーバーjoinで待機
4. エバリュエータの分離実行
   └─ 別スレッドでeval_fnを実行（クラスタからevaluatorを分離）
```

### フローチャート

```mermaid
flowchart TD
    A[開始] --> B{cluster_specあり?}
    B -->|No| C[ローカル実行]
    B -->|Yes| D{モード判定}
    D -->|STANDALONE_CLIENT| E{タスクタイプ}
    D -->|INDEPENDENT_WORKER| F[全ノードでTFサーバー起動]
    E -->|client| G{between-graph?}
    E -->|非client| H[TFサーバー起動+join]
    G -->|Yes| I[ワーカースレッド並列実行]
    G -->|No| J[単一スレッドで実行]
    F --> K{タスクタイプ}
    K -->|chief/worker| L[worker_fn実行]
    K -->|evaluator| M[eval_fn実行]
    K -->|ps| N[サーバーjoin待機]
    C --> O[終了]
    I --> O
    J --> O
    L --> O
    M --> O
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-72-01 | チーフノード判定 | cluster_specにchiefジョブがなければworker:0がchief | cluster_specあり |
| BR-72-02 | エバリュエータ分離 | evaluatorタスクはクラスタから分離される | evaluatorジョブ存在時 |
| BR-72-03 | TF_CONFIG優先 | cluster_spec引数よりTF_CONFIGの値が優先（cluster_spec未指定時） | TF_CONFIG設定時 |
| BR-72-04 | サーバー再起動防止 | 同一スレッドでのサーバー再起動は既存サーバーを再利用 | サーバー既存時 |
| BR-72-05 | デバイスフィルタ清除 | 標準サーバー起動時にstrategy固有のデバイスフィルタを除去 | 非クライアントタスク |

### 計算ロジック

ワーカー数の計算: `num_workers = len(workers) + len(chiefs)`

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ValueError | 設定不正 | TF_CONFIGにclusterまたはtaskキーがない | TF_CONFIGを正しく設定 |
| ValueError | モード不正 | 未知のCoordinatorModeが指定された | STANDALONE_CLIENTまたはINDEPENDENT_WORKERを使用 |
| ValueError | タスク不正 | 未知のtask_typeが指定された | 有効なタスクタイプを使用 |
| ValueError | コンテキスト違反 | worker_fn内で再度distribute_coordinatorを呼び出し | ネスト呼び出しを避ける |

### リトライ仕様

自動リトライ機能は提供されない。

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

ワーカー間の同期は_Barrierクラスにより実現される。全ワーカーがバリアに到達するまでブロックする。

## パフォーマンス要件

- ワーカー間のバリア同期はO(n)の時間複雑度（nはワーカー数）
- between-graphモードでは各ワーカーが独立スレッドで動作し、並列性を確保

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

- gRPC通信の暗号化はrpc_layer設定で制御
- TF_CONFIG環境変数にはクラスタのホスト名・ポート情報が含まれるため、適切なアクセス制御が必要

## 備考

- STANDALONE_CLIENTモードはクライアントがリモートサーバーに接続するモデル
- INDEPENDENT_WORKERモードは各ワーカーが独立して動作するモデル
- run_standard_tensorflow_serverはTF_CONFIGから標準サーバーを起動する便利関数

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | _TaskTypeクラス（36-41行目）、CoordinatorModeクラス（46-57行目）を理解する |
| 1-2 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | _Barrierクラス（60-88行目）の同期メカニズムを理解する |
| 1-3 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | _WorkerContextクラス（99-319行目）の設定情報管理を理解する |

**読解のコツ**: _WorkerContextはPythonのコンテキストマネージャ（__enter__/__exit__）として実装されており、with文で使用される。distribute_coordinator_contextモジュールを通じてグローバルにアクセス可能。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | run_distribute_coordinator関数（627-873行目）が主要エントリーポイント |

**主要処理フロー**:
1. **748-768行目**: TF_CONFIGからクラスタ構成・タスク情報を解決
2. **759-768行目**: cluster_specが未指定時、strategy.extended._cluster_resolverから解決
3. **781-788行目**: ローカルモード（cluster_specなし）の処理
4. **791-822行目**: STANDALONE_CLIENTモードの処理
5. **823-872行目**: INDEPENDENT_WORKERモードの処理

#### Step 3: 内部実行関数を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | _run_single_worker関数（322-356行目）の実行コンテキスト管理 |
| 3-2 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | _run_between_graph_client関数（450-490行目）のスレッド管理 |
| 3-3 | distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | _run_std_server関数（381-447行目）のTFサーバー起動 |

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

```
run_distribute_coordinator()
    │
    ├─ [ローカルモード]
    │   └─ _run_single_worker(worker_fn)
    │       └─ _WorkerContext.__enter__() → worker_fn(strategy) → __exit__()
    │
    ├─ [STANDALONE_CLIENT / client]
    │   ├─ _run_between_graph_client()
    │   │   ├─ threading.Thread(_run_single_worker) × N (workers)
    │   │   └─ threading.Thread(_run_single_worker) × 1 (evaluator)
    │   └─ _run_in_graph_client()
    │       └─ _run_single_worker(worker_fn)
    │
    ├─ [STANDALONE_CLIENT / 非client]
    │   └─ _run_std_server() → server.join()
    │
    └─ [INDEPENDENT_WORKER]
        ├─ _run_std_server()
        ├─ _run_single_worker() (chief/worker)
        └─ server.join() (ps)
```

### データフロー図

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

TF_CONFIG ──────▶  run_distribute_coordinator()  ──────▶  worker_fn戻り値
cluster_spec ──▶       ├─ クラスタ解決                       / None
worker_fn ─────▶       ├─ モード判定
strategy ──────▶       ├─ サーバー起動
                       └─ ワーカー実行
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| distribute_coordinator.py | `tensorflow/python/distribute/distribute_coordinator.py` | ソース | メイン実装（全機能） |
| distribute_coordinator_context.py | `tensorflow/python/distribute/distribute_coordinator_context.py` | ソース | ワーカーコンテキストのグローバル管理 |
| multi_worker_util.py | `tensorflow/python/distribute/multi_worker_util.py` | ソース | ClusterSpec正規化ユーティリティ |
| server_lib.py | `tensorflow/python/training/server_lib.py` | ソース | TensorFlowサーバー実装 |
| coordinator.py | `tensorflow/python/training/coordinator.py` | ソース | スレッドコーディネーター |
| monitored_session.py | `tensorflow/python/training/monitored_session.py` | ソース | セッション作成 |
