Merge branch 'develop' of github.com:PaddlePaddle/GraphNet into develop

Author: lixinqi

docs/CONTRIBUTE_TUTORIAL.md

@@ -255,35 +255,22 @@ python -m graph_net.config \
     --email "[email protected]"
 ```
 
-2. **Package the graph**
+2. **Commit the changes**
 
+Move the new sample to **samples** directory and commit.
 ```bash
-python -m graph_net.pack --output /path/to/output.zip --clear-after-pack True
-```
-
-This API:
-
-a. Packages all files under `$GRAPH_NET_EXTRACT_WORKSPACE` into `/path/to/output.zip` (You can set it to `GraphNet/samples`)
-
-b. Clears the workspace if `--clear-after-pack` is `True`
-
-Note: If third-party ops are used, contributors must include them manually in the package. As long as `validate` passes, no specific folder structure is required.
-
-3. **Commit the changes**
-
-Move the packaged computational graph in the previous step to **samples** directory and commit.
-```bash
-git add <the packaged computational graph>
+git add <the new sample>
 git commit -m "Description"
 ```
+Note: If third-party ops are used, contributors must include them manually in the package.
 
-4. **Push the branch to your fork**
+3. **Push the branch to your fork**
 
 ```bash
 git push origin feature/your-branch-name
 ```
 
-5. **Submit a Pull Request**
+4. **Submit a Pull Request**
 
 > **Note**: For clarity and maintainability, each PR should follow the Single Responsibility Principle (SRP). Submit only a single graph or a focused feature improvement per PR. For example, if you both update extraction logic and collect multiple models, each graph and each method update should be a separate PR.
 

docs/CONTRIBUTE_TUTORIAL_cn.md

@@ -247,32 +247,22 @@ python -m graph_net.config \
     --username "john_doe" \
     --email "[email protected]"
 ```
-2. **打包计算图**
 
-```bash
-python -m graph_net.pack --output /path/to/output.zip --clear-after-pack True
-```
-该API的功能为：
-
- a. 打包`$GRAPH_NET_EXTRACT_WORKSPACE`下的所有文件到`/path/to/output.zip` （可以设置到`GraphNet/samples`）
-   
- b. 若`--clear-after-pack`为`True`，则打包后清空`$GRAPH_NET_EXTRACT_WORKSPACE`
+2. **提交修改**
 
-请注意，如果有第三方算子，需要贡献者自行打包到计算图压缩包内。目前没有特别规定存放的目录结构，但只要通过了validate环节，就可以达到验收标准。
-
-3. **提交修改**
-
-移动上一步打包完成的计算图压缩包到**samples**目录，然后提交。
+移动新增的计算图样本到**samples**目录，然后提交。
 ```bash
-git add <计算图压缩包>
+git add <新计算图样本>
 git commit -m "描述"
 ```
-4. **推送分支到远程**（你的 Fork 仓库）
+请注意，如果有第三方算子，需要贡献者自行打包到计算图压缩包内。
+
+3. **推送分支到远程**（你的 Fork 仓库）
 
 ```bash
 git push origin feature/your-branch-name
 ```
-5. **提交 Pull Request**
+4. **提交 Pull Request**
 
 > **注意**：为方便管理，每个PR应遵守Single Responsibility Principle (SRP)原则，**仅新增单一份计算图、或聚焦于单一功能改进**，避免将多个修改混合提交。例如，如果您修改了抓取方法，然后为支持某类模型收集了数据，那么其中每份单个模型的计算图、修改的新一份抓取方法，都应打开为独立的PR。
 

graph_net/sample_pass/agent_unittest_generator.py

@@ -224,7 +224,7 @@ def __init__(
         self.output_dir = Path(output_dir)
         self.device = self._choose_device(device)
         self.generate_main = generate_main
-        self.try_run = try_run and generate_main
+        self.try_run = try_run
         self.data_input_predicator = self._make_data_input_predicator(
             data_input_predicator_filepath, data_input_predicator_class_name
         )
@@ -244,20 +244,26 @@ def generate(self):
             input_tensor_metas,
             weight_tensor_metas,
         ) = self._get_input_and_weight_tensor_metas(input_arg_names, weight_arg_names)
-        graph_module_desc = GraphModuleDescriptor(
-            device=self.device,
-            generate_main=self.generate_main,
-            model_name=model_name,
-            input_arg_names=input_arg_names,
-            input_tensor_metas=input_tensor_metas,
-            weight_arg_names=weight_arg_names,
-            weight_tensor_metas=weight_tensor_metas,
-            forward_body=self._get_forward_body(
-                graph_module, input_arg_names, weight_arg_names
-            ),
-        )
-        unittest = self._render_template(graph_module_desc)
-        if self._try_to_run_unittest(unittest):
+
+        def _generate_unittest(generate_main):
+            graph_module_desc = GraphModuleDescriptor(
+                device=self.device,
+                generate_main=generate_main,
+                model_name=model_name,
+                input_arg_names=input_arg_names,
+                input_tensor_metas=input_tensor_metas,
+                weight_arg_names=weight_arg_names,
+                weight_tensor_metas=weight_tensor_metas,
+                forward_body=self._get_forward_body(
+                    graph_module, input_arg_names, weight_arg_names
+                ),
+            )
+            return self._render_template(graph_module_desc)
+
+        # Generate unittest with main for try-run.
+        unittest_for_try_run = _generate_unittest(generate_main=self.try_run)
+        if self._try_to_run_unittest(unittest_for_try_run):
+            unittest = _generate_unittest(generate_main=self.generate_main)
             self._write_to_file(unittest, self.output_dir)
 
     def _choose_device(self, device) -> str:

graph_net/sample_pass/fusible_subgraph_ranges_generator.py

@@ -0,0 +1,254 @@
+from graph_net.sample_pass.sample_pass import SamplePass
+from graph_net.sample_pass.resumable_sample_pass_mixin import ResumableSamplePassMixin
+from pathlib import Path
+import json
+from itertools import groupby
+from dataclasses import dataclass
+
+
+class FusibleSubgraphRangesGenerator(SamplePass, ResumableSamplePassMixin):
+    def __init__(self, config):
+        super().__init__(config)
+
+    def declare_config(
+        self,
+        model_path_prefix: str,
+        output_dir: str,
+        input_json_file_name: str,
+        resume: bool = False,
+        limits_handled_models: int = None,
+        output_json_file_name: str = "fusible_subgraph_ranges.json",
+    ):
+        pass
+
+    def __call__(self, rel_model_path: str):
+        self.resumable_handle_sample(rel_model_path)
+
+    def sample_handled(self, rel_model_path: str) -> bool:
+        file_name = self.config["output_json_file_name"]
+        return self.naive_sample_handled(rel_model_path, search_file_name=file_name)
+
+    def resume(self, rel_model_path: str):
+        analyzer = self._make_analyzer(rel_model_path)
+        output_obj = {
+            "subgraph_ranges": analyzer.analyze(),
+        }
+        self._save_output(rel_model_path, output_obj)
+
+    def _save_output(self, rel_model_path, output_obj):
+        output_json = json.dumps(output_obj, indent=4)
+        output_dir_path = Path(self.config["output_dir"]) / rel_model_path
+        output_dir_path.mkdir(parents=True, exist_ok=True)
+        output_file_path = output_dir_path / self.config["output_json_file_name"]
+        output_file_path.write_text(output_json)
+
+    def _make_analyzer(self, rel_model_path: str):
+        model_path = (
+            Path(self.config["model_path_prefix"])
+            / rel_model_path
+            / self.config["input_json_file_name"]
+        )
+        json_ctx = self._make_json_ctx(model_path)
+        return FusibleSubgraphRangesAnalyzer(
+            num_subgraph_kernels_list=self._get_num_subgraph_kernels_list(json_ctx),
+            num_subgraph_ops_list=self._get_num_subgraph_ops_list(json_ctx),
+            start_offset_in_original_graph=self._get_start_offset_in_original_graph(
+                json_ctx
+            ),
+        )
+
+    def _get_start_offset_in_original_graph(self, json_ctx):
+        return json_ctx["start_offset_in_original_graph"]
+
+    def _get_num_subgraph_kernels_list(self, json_ctx):
+        return json_ctx["num_subgraph_kernels"]
+
+    def _get_num_subgraph_ops_list(self, json_ctx):
+        return json_ctx["num_subgraph_ops"]
+
+    def _make_json_ctx(self, model_path: Path):
+        obj = json.loads(model_path.read_text())
+        assert len(obj["num_subgraph_kernels"]) == len(obj["num_subgraph_ops"])
+        return obj
+
+
+class FusibleSubgraphRangesAnalyzer:
+    def __init__(
+        self,
+        num_subgraph_kernels_list: list[int],
+        num_subgraph_ops_list: list[int],
+        start_offset_in_original_graph: int,
+    ):
+        assert len(num_subgraph_kernels_list) == len(num_subgraph_ops_list)
+        self.num_subgraph_kernels_list = num_subgraph_kernels_list
+        self.num_subgraph_ops_list = num_subgraph_ops_list
+        self.start_offset_in_original_graph = start_offset_in_original_graph
+
+    def analyze(self):
+        analysis_ctx = self._make_analysis_ctx()
+        num_kernels_and_num_ops_list = analysis_ctx.num_kernels_and_num_ops_list
+        # The tail num_kernels equals the head num_kernels for each num_ops_list
+        naive_proposal_fused_num_ops_lists = [
+            sorted(set(num_ops_list))
+            for _, num_ops_list in num_kernels_and_num_ops_list
+            if len(set(num_ops_list)) > 1
+        ]
+        proposal_fused_num_ops_lists = self._merge_all_decreasing_num_ops_lists(
+            analysis_ctx, naive_proposal_fused_num_ops_lists
+        )
+        return self._create_subgraph_ranges_from_proposal(
+            analysis_ctx,
+            proposal_fused_num_ops_lists,
+        )
+
+    def _merge_all_decreasing_num_ops_lists(self, analysis_ctx, num_ops_lists):
+        dead_loop_detect_cnt = 0
+        kLimit = 99999
+        while True:
+            last_len_num_ops_lists = len(num_ops_lists)
+            num_ops_lists = self._merge_one_decreasing_num_ops_lists(
+                analysis_ctx, num_ops_lists
+            )
+            assert last_len_num_ops_lists >= len(num_ops_lists)
+            if last_len_num_ops_lists == len(num_ops_lists):
+                break
+            dead_loop_detect_cnt += 1
+            assert dead_loop_detect_cnt < kLimit, f"{dead_loop_detect_cnt=}"
+        return num_ops_lists
+
+    def _merge_one_decreasing_num_ops_lists(self, analysis_ctx, num_ops_lists):
+        merge_pos = self._detect_mergable_decreasing_position(
+            analysis_ctx, num_ops_lists
+        )
+        if merge_pos is None:
+            return num_ops_lists
+        assert merge_pos >= 0
+        assert merge_pos < len(num_ops_lists) - 1
+        return [
+            *num_ops_lists[:merge_pos],
+            [*num_ops_lists[merge_pos], *num_ops_lists[merge_pos + 1]],
+            *num_ops_lists[merge_pos + 2 :],
+        ]
+
+    def _detect_mergable_decreasing_position(self, analysis_ctx, num_ops_lists):
+        def get_cur_tail_num_kernels(i):
+            return analysis_ctx.num_kernels4num_ops(num_ops_lists[i][-1])
+
+        def get_next_head_num_kernels(i):
+            return analysis_ctx.num_kernels4num_ops(num_ops_lists[i + 1][0])
+
+        for i in range(len(num_ops_lists) - 1):
+            assert len(num_ops_lists[i]) > 1
+            if get_cur_tail_num_kernels(i) >= get_next_head_num_kernels(i):
+                return i
+        return None
+
+    def _create_subgraph_ranges_from_proposal(
+        self, analysis_ctx, proposal_fused_num_ops_lists
+    ):
+        # filter valid num_ops_list
+
+        def is_a_range(int_list):
+            assert len(int_list) > 1
+            return (int_list[-1] + 1) - int_list[0] == len(int_list)
+
+        def have_any_increasing(num_ops_list: list[int]):
+            for i, cur_num_ops in enumerate(num_ops_list):
+                if i == 0:
+                    continue
+                cur_num_kernels = analysis_ctx.num_kernels4num_ops(cur_num_ops)
+                last_num_kernels = analysis_ctx.num_kernels4num_ops(num_ops_list[i - 1])
+                if cur_num_kernels > last_num_kernels:
+                    return True
+            return False
+
+        def head_eq_tail(num_ops_list: list[int]):
+            return analysis_ctx.num_kernels4num_ops(
+                num_ops_list[0]
+            ) == analysis_ctx.num_kernels4num_ops(num_ops_list[-1])
+
+        def head_gt_tail(num_ops_list: list[int]):
+            return analysis_ctx.num_kernels4num_ops(
+                num_ops_list[0]
+            ) > analysis_ctx.num_kernels4num_ops(num_ops_list[-1])
+
+        def valid_fused_ops(num_ops_list: list[int]):
+            if head_gt_tail(num_ops_list):
+                return True
+            if head_eq_tail(num_ops_list):
+                return not have_any_increasing(num_ops_list)
+            return False
+
+        proposal_fused_num_ops_lists = [
+            sorted(set(num_ops_list)) for num_ops_list in proposal_fused_num_ops_lists
+        ]
+        num_ops_lists = [
+            num_ops_list
+            for num_ops_list in proposal_fused_num_ops_lists
+            if len(num_ops_list) > 1
+            if is_a_range(num_ops_list)
+            if valid_fused_ops(num_ops_list)
+        ]
+        fusible_subgraph_ranges = [
+            (start, end)
+            for num_ops_list in num_ops_lists
+            for start in [num_ops_list[0] - 1]
+            for end in [num_ops_list[-1]]
+        ]
+
+        # sorted by `start`
+        def range_sort_key(pair):
+            start, end = pair
+            # smaller `start` first
+            # bigger `end` first
+            return (start, -end)
+
+        fusible_subgraph_ranges = sorted(fusible_subgraph_ranges, key=range_sort_key)
+        # remove shadowed
+        fusible_subgraph_ranges = [
+            fusible_subgraph_ranges[i]
+            for i in range(len(fusible_subgraph_ranges))
+            if i == 0
+            or (fusible_subgraph_ranges[i][0] >= fusible_subgraph_ranges[i - 1][1])
+        ]
+        return fusible_subgraph_ranges
+
+    def _make_analysis_ctx(self):
+        return AnalysisContext(
+            num_kernels_and_num_ops_list=self._make_num_kernels_and_num_ops_list(),
+            num_ops2num_kernels=self._make_num_ops2num_kernels(),
+        )
+
+    def _make_num_ops2num_kernels(self):
+        return dict(zip(self.num_subgraph_ops_list, self.num_subgraph_kernels_list))
+
+    def _make_num_kernels_and_num_ops_list(self):
+        num_kernels_and_num_ops = zip(
+            self.num_subgraph_kernels_list,
+            self.num_subgraph_ops_list,
+        )
+
+        def get_num_kernels(pair):
+            return pair[0]
+
+        def get_num_ops(pair):
+            return pair[1]
+
+        num_kernels_and_num_ops = sorted(num_kernels_and_num_ops, key=get_num_ops)
+        grouped_num_kernels_and_num_ops = groupby(
+            num_kernels_and_num_ops, key=get_num_kernels
+        )
+        num_kernels_and_num_ops_list = [
+            (num_kernels, [num_ops for _, num_ops in group])
+            for num_kernels, group in grouped_num_kernels_and_num_ops
+        ]
+        return num_kernels_and_num_ops_list
+
+
+@dataclass
+class AnalysisContext:
+    num_kernels_and_num_ops_list: list[(int, list[int])]
+    num_ops2num_kernels: dict[int, int]
+
+    def num_kernels4num_ops(self, num_ops: int):
+        return self.num_ops2num_kernels[num_ops]