# 機能設計書 9-データフロー操作（Data Flow Operations）

## 概要

本ドキュメントは、TensorFlowのコアフレームワークにおけるデータフロー操作（Data Flow Operations）機能の設計を記述する。キュー・スタック・TensorArray・バリアなどのデータフロー制御を提供する。

### 本機能の処理概要

データフロー操作機能は、計算グラフ内でのデータの動的な流れを制御するための構造を提供する。DynamicPartition/DynamicStitchによるデータの動的分割・結合、キュー（FIFOQueue, PaddingFIFOQueue等）によるプロデューサ-コンシューマパターン、TensorArrayによる動的長さの配列操作、Barrier/StagingAreaによる同期制御など、非同期・動的なデータパイプラインの構築を可能にする。

**業務上の目的・背景**：大規模な機械学習パイプラインでは、データの前処理・バッチ生成・モデル学習を非同期に実行する必要がある。キュー操作によるデータパイプラインの構築、DynamicPartitionによるデータの条件付き分割、TensorArrayによるRNNの動的展開など、動的なデータフロー制御が必要となる。

**機能の利用シーン**：(1) DynamicPartitionによるラベル別データ分割、(2) DynamicStitchによる分割データの再結合、(3) FIFOQueueによる非同期データパイプライン、(4) TensorArrayによるwhileloop内のテンソル蓄積、(5) StagingAreaによるGPUフィーディング。

**主要な処理内容**：
1. DynamicPartition: テンソルの動的分割（パーティションインデックスに基づく）
2. DynamicStitch / ParallelDynamicStitch: 分割テンソルの再結合
3. FIFOQueue / PaddingFIFOQueue / PriorityQueue / RandomShuffleQueue: 各種キュー操作
4. QueueEnqueue / QueueDequeue / QueueClose / QueueSize: キュー操作API
5. TensorArray: 動的長さテンソル配列
6. Barrier: 同期バリア
7. StagingArea: ステージングエリア
8. GetSessionHandle / GetSessionTensor / DeleteSessionTensor: セッションテンソル管理

**関連システム・外部連携**：tf.data Dataset API（機能No.44）が後継機能として多くのキュー操作を置き換える。制御フロー操作（機能No.6）のwhile_loopと組み合わせてTensorArrayが使用される。

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

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能はコアAPIのため直接的な画面関連はなし |

## 機能種別

データ連携（データフロー制御・非同期パイプライン）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| data | tf.Tensor | DynamicPartition時 | 分割対象テンソル | 任意の型 |
| partitions | tf.Tensor(int32) | DynamicPartition時 | パーティションインデックス | [0, num_partitions)の範囲 |
| num_partitions | int | DynamicPartition時 | パーティション数 | 正の整数 |
| indices | tf.Tensor(int32) | DynamicStitch時 | 結合インデックス | - |
| dtypes | list[DType] | キュー時 | キュー要素のデータ型リスト | サポートされる型 |
| shapes | list[TensorShape] | キュー時 | キュー要素の形状リスト | - |
| capacity | int | キュー時 | キューの最大容量 | 正の整数 |

### 入力データソース

Python API呼び出し時の引数として入力される。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| outputs | list[tf.Tensor] | DynamicPartition: num_partitions個のテンソル |
| merged | tf.Tensor | DynamicStitch: 結合されたテンソル |
| handle | resource | キュー: キューハンドル |
| components | list[tf.Tensor] | QueueDequeue: デキューされた要素 |

### 出力先

呼び出し元のPythonコードに返却される。キュー操作の場合はキューリソースを通じてデータが受け渡される。

## 処理フロー

### 処理シーケンス

```
1. データフロー構造の作成（キュー/TensorArray/StagingArea）
   └─ リソースハンドルの生成
2. データの投入
   └─ Enqueue / TensorArrayWrite / DynamicPartition
3. データの取得
   └─ Dequeue / TensorArrayRead / DynamicStitch
4. リソースの解放
   └─ QueueClose / リソース破棄
```

### フローチャート

```mermaid
flowchart TD
    A[データフロー構造作成] --> B{構造種別}
    B -->|DynamicPartition| C[パーティションインデックスで分割]
    B -->|キュー| D[FIFOQueue/等の作成]
    B -->|TensorArray| E[動的配列作成]
    C --> F[num_partitions個の出力テンソル]
    D --> G[Enqueue/Dequeue操作]
    E --> H[Write/Read操作]
    F --> I[結果返却]
    G --> I
    H --> I
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | パーティション前方一致 | DynamicPartitionのdata形状はpartitions形状で始まる | DynamicPartition時 |
| BR-02 | キュー容量制限 | キューがcapacityに達するとEnqueueはブロック | キュー操作時 |
| BR-03 | キュークローズ制約 | クローズ済みキューへのEnqueueはエラー | キュークローズ後 |
| BR-04 | 型一致 | DynamicStitchのすべてのデータテンソルは同一型 | DynamicStitch時 |

### 計算ロジック

- DynamicPartition: data を partitions[i] の値に基づいて num_partitions 個のテンソルに分割
- DynamicStitch: indices と data を使用して、indices[i][j] 番目の位置に data[i][j] を配置

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

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

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

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

該当なし。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| InvalidArgumentError | 形状エラー | data形状がpartitions形状と不一致 | 整合性のある形状を使用する |
| CancelledError | キャンセル | クローズ済みキューへのDequeue | キュー状態を確認する |
| OutOfRangeError | 範囲超過 | 空のクローズ済みキューからのDequeue | データ終了を適切にハンドリングする |

### リトライ仕様

キュー操作はブロッキングであり、timeout_ms引数によりタイムアウトを設定可能。

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

キュー操作は個別にアトミックであるが、複数のキュー操作を跨ぐトランザクションはサポートされない。

## パフォーマンス要件

- DynamicPartition/DynamicStitchは大規模テンソルに対して効率的に実行すること
- キュー操作はプロデューサ-コンシューマ間で最小のオーバーヘッドで実行すること

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

- キューのcapacity設定によるメモリ使用量の制御
- デッドロック防止のためのtimeout設定

## 備考

- tf.data API（機能No.44）がキュー操作の推奨代替手段
- DynamicPartition/DynamicStitchは依然として広く使用されている
- Python側のdata_flow_ops.pyはヘルパー関数（_as_type_list, _as_shape_list, _as_name_list）を提供

---

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

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

### 推奨読解順序

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

データフロー操作で使用されるヘルパー関数と型変換を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | data_flow_ops.py | `tensorflow/python/ops/data_flow_ops.py` | ヘルパー関数群 |

**読解のコツ**: data_flow_ops.pyの冒頭にある `_as_type_list`, `_as_shape_list`, `_as_name_list` はキュー操作の引数を正規化するヘルパー関数である。

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

Python側のヘルパー関数とC++ op定義を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | data_flow_ops.py | `tensorflow/python/ops/data_flow_ops.py` | ヘルパー関数 |

**主要処理フロー**:
1. **45-53行目**: `_as_type_list` - dtypesをリストに正規化
2. **56-86行目**: `_as_shape_list` - shapesをTensorShapeリストに正規化
3. **89-98行目**: `_as_name_list` - namesをリストに正規化（dtypesと長さ一致を検証）

#### Step 3: C++オペレーション定義を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | data_flow_ops.cc | `tensorflow/core/ops/data_flow_ops.cc` | REGISTER_OPによるop定義 |

**主要処理フロー**:
- **29-42行目**: `DequeueManyV2Shape` - DequeueMany操作の形状推論
- **48行目**: `DynamicPartition` op登録 - テンソルの動的分割
- **91行目以降**: `DynamicStitch` shape関数 - 動的結合の形状推論

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

```
tf.dynamic_partition(data, partitions, num_partitions)
    |
    +-- gen_data_flow_ops.dynamic_partition()
            |
            +-- C++ DynamicPartition kernel
                    |
                    +-- DynamicPartitionShapeFn (形状推論)
                    +-- partitions値に基づいてdataを分割

tf.dynamic_stitch(indices, data)
    |
    +-- gen_data_flow_ops.parallel_dynamic_stitch()
            |
            +-- C++ ParallelDynamicStitch kernel
```

### データフロー図

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

data, partitions   -->  DynamicPartition           -->  outputs[0..N-1]
(テンソル, int32)       (パーティション分割)

indices, data      -->  DynamicStitch              -->  merged tensor
(int32, テンソル)       (インデックス基づく結合)

components         -->  FIFOQueue.enqueue          -->  (キューに格納)
                        FIFOQueue.dequeue          -->  components
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| data_flow_ops.py | `tensorflow/python/ops/data_flow_ops.py` | ソース | Pythonヘルパー関数・キューラッパー |
| data_flow_ops.cc | `tensorflow/core/ops/data_flow_ops.cc` | ソース | C++ op定義 |
| gen_data_flow_ops.py | `tensorflow/python/ops/gen_data_flow_ops.py` | 自動生成 | Python バインディング |
| tensor_array_ops.py | `tensorflow/python/ops/tensor_array_ops.py` | ソース | TensorArray操作 |
| common_shape_fns.h | `tensorflow/core/framework/common_shape_fns.h` | ヘッダ | 形状推論関数 |
