You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tvm.apache.org by GitBox <gi...@apache.org> on 2022/05/17 06:02:35 UTC
[GitHub] [tvm] huajsj opened a new pull request, #11334: [Runtime][PipelineExecutor] Add graph manually splitting example into the unit test.
huajsj opened a new pull request, #11334:
URL: https://github.com/apache/tvm/pull/11334
Current unit test create 3 seperate module then re-connect them to run the pipeline executor. And this is not a real use case for pipeline executor.
Adding a manually graph splitting logic which split a full network into 3 subgraph then run the pipeline executor and verify the result to simulate the real use case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875546489
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
Review Comment:
keep it to compatible with old python version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on PR #11334:
URL: https://github.com/apache/tvm/pull/11334#issuecomment-1130398221
@SebastianBoblestETAS, all review comments addressed, please take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875538130
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
Review Comment:
this not work with some python version, to compatible with old version still keep it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875212280
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,195 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Split graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency of this subgraph node
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var recreate it
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ # If body not let, then reached end of the express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operator of compute graph then split the compute graph into a group subgraph.
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875537165
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
Review Comment:
this is a "integer" compare and not a string comparison. added comments to avoid confusion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875548930
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
+ # such constant may become free varaible due to the constant
Review Comment:
added more detailed comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] SebastianBoblestETAS commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
SebastianBoblestETAS commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875542918
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
Review Comment:
What do you mean with "compatible with old python version"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875528806
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
Review Comment:
the function need to get the key only so we can not remove the ',_' .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on PR #11334:
URL: https://github.com/apache/tvm/pull/11334#issuecomment-1129547988
@SebastianBoblestETAS , please take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] masahi merged pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
masahi merged PR #11334:
URL: https://github.com/apache/tvm/pull/11334
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875551019
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875565772
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875565553
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] SebastianBoblestETAS commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
SebastianBoblestETAS commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r874504485
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,195 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Split graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency of this subgraph node
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var recreate it
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ # If body not let, then reached end of the express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operator of compute graph then split the compute graph into a group subgraph.
Review Comment:
```suggestion
# Enumerate all operators of compute graph then split the compute graph into a group subgraph.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] SebastianBoblestETAS commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
SebastianBoblestETAS commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875531850
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
Review Comment:
I also removed the .items() so this should be ok. But please take this only as a suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] zhaoyang-star commented on pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
zhaoyang-star commented on PR #11334:
URL: https://github.com/apache/tvm/pull/11334#issuecomment-1283603687
Hi @huajsj , Thanks for your contribution. I recently tried to use Pipeline Executor to optimize my inference. Based on `test_pipeline_executor.py` I tried to split ResNet18 to 3 subgraphs and verify the result compared to the normal way. But the result of pipeline diff with the normal way.
```shell
np.testing.assert_allclose(actual.shape, desired.shape)
> np.testing.assert_allclose(actual, desired, rtol=rtol, atol=atol, verbose=True)
E AssertionError:
E Not equal to tolerance rtol=1e-07, atol=1e-07
E
E Mismatched elements: 1000 / 1000 (100%)
E Max absolute difference: 0.0008707
E Max relative difference: 0.7465138
E x: array([[0.000775, 0.001194, 0.000832, 0.001083, 0.00134 , 0.001236,
E 0.001081, 0.00083 , 0.001295, 0.001404, 0.000695, 0.000878,
E 0.000933, 0.000871, 0.00145 , 0.000928, 0.001229, 0.000944,...
E y: array([[0.000947, 0.00105 , 0.000971, 0.001059, 0.001042, 0.001014,
E 0.001078, 0.000846, 0.001069, 0.000998, 0.000913, 0.001075,
E 0.000989, 0.000885, 0.00121 , 0.000972, 0.001071, 0.000932,...
```
I have checked the subgraphs and found they are all right. Could you please help to find the reason of difference between pipeline way and the normal way? Thanks for your kind help.
Please run `apt-get install graphvis` first for visulize subgraph.
```python3
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import pytest
import os
import time
import numpy as np
import tvm
import tvm.testing
from tvm import relay
from tvm.relay import transform, build_module
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
import inspect
tutorial_dir = os.path.dirname(inspect.getfile(lambda: None))
os.sys.path.append(os.path.join(tutorial_dir, "../../../tests/python/relay"))
from test_pipeline_executor import graph_split, reset_cpu_affinity
from tvm.contrib import relay_viz
graph_attr = {"color": "red"}
node_attr = {"color": "blue"}
edge_attr = {"color": "black"}
# VizNode is passed to the callback.
def get_node_attr(node):
if "nn.conv2d" in node.type_name and "NCHW" in node.detail:
return {
"fillcolor": "green",
"style": "filled",
"shape": "box",
}
if "Var" in node.type_name:
return {"shape": "ellipse"}
return {"shape": "box"}
dot_plotter = relay_viz.DotPlotter(
graph_attr=graph_attr,
node_attr=node_attr,
edge_attr=edge_attr,
get_node_attr=get_node_attr)
def get_split_mod():
mod, params = tvm.relay.testing.resnet.get_workload(
batch_size=1, num_classes=1000, num_layers=18)
print("Outputing mod.pdf ...")
viz = relay_viz.RelayVisualizer(
mod,
plotter=dot_plotter,
parser=relay_viz.DotVizParser())
viz.render("mod")
split_conf = [{"op_name": "add", "op_index": 1}, {"op_name": "add", "op_index": 4}]
mods = graph_split(mod["main"], split_conf, params)
return mods, params, (1, 3, 224, 224), mod
def build_relay(mod, params):
dev = tvm.device("llvm", 0)
with tvm.transform.PassContext(opt_level=3):
lib = relay.build_module.build(mod, "llvm", params=params)
module = graph_executor.GraphModule(lib["default"](dev))
return module
def run_relay(module, in_data):
module.set_input("data", in_data)
module.run()
return module.get_output(0).numpy()
def test_pipeline():
if pipeline_executor_build.pipeline_executor_build_enabled():
target_list = tvm.testing.enabled_targets()
print(f"target_list:{target_list}")
for target in target_list:
affinity = os.sched_getaffinity(0)
# Get the three pipeline modules here.
(mod1, mod2, mod3), params, dshape, ori_mod = get_split_mod()
print("Outputing mod1.pdf ...")
viz = relay_viz.RelayVisualizer(
mod1,
plotter=dot_plotter,
parser=relay_viz.DotVizParser())
viz.render("mod1")
print("Outputing mod2.pdf ...")
viz = relay_viz.RelayVisualizer(
mod2,
plotter=dot_plotter,
parser=relay_viz.DotVizParser())
viz.render("mod2")
print("Outputing mod3.pdf ...")
viz = relay_viz.RelayVisualizer(
mod3,
plotter=dot_plotter,
parser=relay_viz.DotVizParser())
viz.render("mod3")
# Prepare batch data for pipeline computation.
datas = []
for i in range(5):
datas.append(np.full(dshape, 3 + i).astype("float32"))
pipe_config = pipeline_executor_build.PipelineConfig()
# The pipeline input named "data_a" will be connected to a input named "data"
# of mod1.
pipe_config["input"]["data_a"].connect(pipe_config[mod1]["input"]["data"])
# The mod1 output[0] will be connected to a input named "data_n_0" of mod2.
pipe_config[mod1]["output"][0].connect(pipe_config[mod2]["input"]["data_n_0"])
# The mod2 output[2] will be connected to a input named "data_n_1" of mod3.
pipe_config[mod2]["output"][0].connect(pipe_config[mod3]["input"]["data_n_1"])
# The mod3 output[0] will be connected to pipeline output[0].
pipe_config[mod3]["output"][0].connect(pipe_config["output"]["0"])
print(f"pipe_config:\n {pipe_config}")
# Print configuration (print(pipe_config)), the result looks like following.
#
# Params
#
# Inputs
# |data_a: mod0:data
#
# output
# |output(0) : mod2.output(0)
#
# connections
# |mod0.output(0)-> mod1.data_n_0
# |mod1.output(0)-> mod2.data_n_1
# Set other parameters.
pipe_config[mod1].target = target[0]
pipe_config[mod1].dev = target[1]
pipe_config[mod1].cpu_affinity = "0"
pipe_config[mod1].fcompile = _cc.create_shared
pipe_config[mod2].target = "llvm"
pipe_config[mod2].dev = tvm.cpu(0)
pipe_config[mod2].cpu_affinity = "1"
pipe_config[mod3].target = "llvm"
pipe_config[mod3].dev = tvm.cpu(0)
pipe_config[mod3].cpu_affinity = "2"
# Checking the configuration of modules dependency.
mconfig = pipe_config.get_config()
with open('module_connection.config', 'w') as f:
f.write(f'module_connection config:\n{mconfig["module_connection"]}')
# Build and create a pipeline module.
with tvm.transform.PassContext(opt_level=3):
pipeline_mod_factory = pipeline_executor_build.build(pipe_config)
# Export the parameter configuration to a file.
directory_path = tvm.contrib.utils.tempdir().temp_dir
# If the directory does not exist, create it.
if not os.path.exists(directory_path):
os.makedirs(directory_path)
config_file_name = pipeline_mod_factory.export_library(directory_path)
# Use the output of build to create and initialize PipelineModule.
pipeline_module = pipeline_executor.PipelineModule(pipeline_mod_factory)
assert pipeline_module
# Use the import function to create and initialize PipelineModule.
pipeline_module_test = pipeline_executor.PipelineModule.load_library(config_file_name)
assert pipeline_module_test.num_outputs == 1
input_map = pipeline_module_test.get_input_pipeline_map("data_a")
assert input_map[0] == "0" and input_map[1] == "data"
# Build the original mod
normal_outputs = []
module = build_relay(ori_mod, params)
for round in range(0, len(datas)):
data = datas[round]
# Setting the input data into the pipeline executor.
pipeline_module_test.set_input("data_a", tvm.nd.array(data))
input_map = pipeline_module_test.get_input_pipeline_map("data_a")
# Checking whether the input setting of the first runtime is successful.
# The input of the rest of runtime will go into a queue and we can not check
# these input data here.
if input_map[0] == "0":
input_data = pipeline_module_test.get_input("data_a")
tvm.testing.assert_allclose(data, input_data.numpy())
assert pipeline_module_test.num_inputs == 1
# Running the pipeline executor in the pipeline mode.
pipeline_module_test.run()
# Running the original mod.
normal_outputs.append(run_relay(module, data))
for k in range(0, len(datas)):
statistic_time = 0
outputs = pipeline_module_test.get_output()
while len(outputs) == 0:
outputs = pipeline_module_test.get_output()
statistic_time = statistic_time + 1
# Setting the timeout to 10 seconds.
assert statistic_time < 5
time.sleep(1)
# np.savetxt("baseline.out", normal_outputs[k])
# np.savetxt("pipeline.out", outputs[0].numpy())
for i in range(len(outputs)):
tvm.testing.assert_allclose(normal_outputs[i], outputs[i].numpy())
assert pipeline_module_test.num_executing_pipeline == round + 1
# Reset the cpu affinity after a test.
# reset_cpu_affinity(affinity)
if __name__ == "__main__":
pytest.main([__file__])
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] SebastianBoblestETAS commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
SebastianBoblestETAS commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r876658319
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -73,7 +71,7 @@ def merge_constant_expr(constant_expr, expr):
)
def _recursion(anf, pipeline_mods, split_conf, constant_expr):
- # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # Enumurate all operators of compute graph, then split the compute graph into a group of
Review Comment:
```suggestion
# Enumerate all operators of compute graph, then split the compute graph into a group of
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -123,13 +125,13 @@ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
)
snode_dep.pop()
dep_vars = get_dep_var(snode_dep)
- # When the nodes of current subgraph are the depedency node of other
+ # When the nodes of the current subgraph are the depedency node of another
# subgraph, we need to set them as the output of current subgraph.
body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
- # when current subgraph use previous subgraph constant,
- # such constant may become free varaible due to the constant
- # not exist, merge the previous constant with current subgraph
- # to avoid such issue.
+ # when the operator of current subgraph uses previous subgraph constant
+ # as the argument of a "relay.expr.call", such constant may become a free
+ # varaible if the constant does not exist in the current subgraph.
Review Comment:
```suggestion
# variable if the constant does not exist in the current subgraph.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875538552
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
Review Comment:
keep it to compatible with old python version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875569135
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875567664
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
Review Comment:
yes, the change work, fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] SebastianBoblestETAS commented on a diff in pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
SebastianBoblestETAS commented on code in PR #11334:
URL: https://github.com/apache/tvm/pull/11334#discussion_r875461735
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
Review Comment:
``
```suggestion
# Enumerate all operators of compute graph, then split the compute graph into a group of
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
Review Comment:
```suggestion
# When the nodes of the current subgraph are the depedency node of another
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
Review Comment:
```suggestion
# record the constant expr to make sure all subgraphs can find correct constant.
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
+ # such constant may become free varaible due to the constant
Review Comment:
```suggestion
# such constant may become a free variable if the constant does not exist.
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
+ # such constant may become free varaible due to the constant
+ # not exist, merge the previous constant with current subgraph
+ # to avoid such issue.
+ if constant_expr:
+ ann = merge_constant_expr(constant_expr, ann)
+ ann = run_opt_pass(ann, transform.ToGraphNormalForm())
+ mod = tvm.IRModule.from_expr(ann)
+ pipeline_mods.insert(0, mod)
+ # Return the last node of the current subgraph.
+ return tvm.relay.expr.Let(anf.var, value, body)
+ return tvm.relay.expr.Let(
+ anf.var,
+ value,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ )
+ else:
+ return anf
+
+ snode_dep = [{"nodes": {}, "ref_nodes": {}}]
+ pipeline_mods = []
+ operator_index_map = {}
+ # Used to tracking new input which caused by graph splitting.
+ new_input_idx = 0
+ constant_expr = None
+ subgraph_split_conf = split_conf.copy()
+ # Binding the parameters.
+ if params:
+ expr = build_module.bind_params_by_name(expr, params)
+ anf = run_opt_pass(expr, transform.ToANormalForm())
+ anf = run_opt_pass(anf, transform.InferType())
+ ann = _recursion(
+ anf,
+ pipeline_mods,
+ subgraph_split_conf,
+ constant_expr,
+ )
+ ann = run_opt_pass(ann.body, transform.ToGraphNormalForm())
+ mod = tvm.IRModule.from_expr(ann)
+ pipeline_mods.insert(0, mod)
+ return pipeline_mods
+
+
+def get_network():
+ # Get a list of modules representing subgraphs.
+ mods = []
+ dshape = (3, 3)
+ data = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ data21 = relay.var("data_1", relay.TensorType(dshape, "float32"))
+ data_net1_output_1 = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ data_net1_output_2 = relay.var("data_1", relay.TensorType(dshape, "float32"))
+ data_net2_output_1 = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ mvalue1 = np.full((1), 1).astype("float32")
+ mvalue2 = np.full((1), 2).astype("float32")
+ mvalue3 = np.full((1), 3).astype("float32")
+ mv1 = relay.Constant(tvm.nd.array(mvalue1))
+ mv2 = relay.Constant(tvm.nd.array(mvalue2))
+ mv3 = relay.Constant(tvm.nd.array(mvalue3))
+ # There are three outputs in the first model.
+ net1_output1 = relay.add(data, mv1)
+ net1_output2 = relay.subtract(data, mv2)
+ net1_output3 = relay.concatenate((net1_output1, net1_output2), axis=0)
+ (net1_output3, _) = relay.split(net1_output3, indices_or_sections=2, axis=0)
+ net1_output3 = relay.add(net1_output3, mv2)
+ # The second model use output named net1_output1 of the first model as the first input,
+ # the second input of the second model is data21.
+ net2 = relay.add(net1_output3, mv2)
+ net2 = relay.add(net2, data21)
+ net2_output = relay.add(net2, mv3)
+ # The third model use the output named net2_output of the second model as the first input
Review Comment:
```suggestion
# The third model uses the output named net2_output of the second model as the first input
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
Review Comment:
```suggestion
return [var for var in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"]]
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
Review Comment:
```suggestion
"""Splitting graph into a list of subgraphs"""
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
Review Comment:
```suggestion
operator_index_map[value.op.name] += 1
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
+ # such constant may become free varaible due to the constant
Review Comment:
I do not fully understand this comment
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
Review Comment:
This could be made a docstring of graph_split.
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
Review Comment:
```suggestion
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
+ # such constant may become free varaible due to the constant
+ # not exist, merge the previous constant with current subgraph
+ # to avoid such issue.
+ if constant_expr:
+ ann = merge_constant_expr(constant_expr, ann)
+ ann = run_opt_pass(ann, transform.ToGraphNormalForm())
+ mod = tvm.IRModule.from_expr(ann)
+ pipeline_mods.insert(0, mod)
+ # Return the last node of the current subgraph.
+ return tvm.relay.expr.Let(anf.var, value, body)
+ return tvm.relay.expr.Let(
+ anf.var,
+ value,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ )
+ else:
+ return anf
+
+ snode_dep = [{"nodes": {}, "ref_nodes": {}}]
+ pipeline_mods = []
+ operator_index_map = {}
+ # Used to tracking new input which caused by graph splitting.
+ new_input_idx = 0
+ constant_expr = None
+ subgraph_split_conf = split_conf.copy()
+ # Binding the parameters.
+ if params:
+ expr = build_module.bind_params_by_name(expr, params)
+ anf = run_opt_pass(expr, transform.ToANormalForm())
+ anf = run_opt_pass(anf, transform.InferType())
+ ann = _recursion(
+ anf,
+ pipeline_mods,
+ subgraph_split_conf,
+ constant_expr,
+ )
+ ann = run_opt_pass(ann.body, transform.ToGraphNormalForm())
+ mod = tvm.IRModule.from_expr(ann)
+ pipeline_mods.insert(0, mod)
+ return pipeline_mods
+
+
+def get_network():
+ # Get a list of modules representing subgraphs.
+ mods = []
+ dshape = (3, 3)
+ data = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ data21 = relay.var("data_1", relay.TensorType(dshape, "float32"))
+ data_net1_output_1 = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ data_net1_output_2 = relay.var("data_1", relay.TensorType(dshape, "float32"))
+ data_net2_output_1 = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ mvalue1 = np.full((1), 1).astype("float32")
+ mvalue2 = np.full((1), 2).astype("float32")
+ mvalue3 = np.full((1), 3).astype("float32")
+ mv1 = relay.Constant(tvm.nd.array(mvalue1))
+ mv2 = relay.Constant(tvm.nd.array(mvalue2))
+ mv3 = relay.Constant(tvm.nd.array(mvalue3))
+ # There are three outputs in the first model.
+ net1_output1 = relay.add(data, mv1)
+ net1_output2 = relay.subtract(data, mv2)
+ net1_output3 = relay.concatenate((net1_output1, net1_output2), axis=0)
+ (net1_output3, _) = relay.split(net1_output3, indices_or_sections=2, axis=0)
+ net1_output3 = relay.add(net1_output3, mv2)
+ # The second model use output named net1_output1 of the first model as the first input,
+ # the second input of the second model is data21.
+ net2 = relay.add(net1_output3, mv2)
+ net2 = relay.add(net2, data21)
+ net2_output = relay.add(net2, mv3)
+ # The third model use the output named net2_output of the second model as the first input
+ # and use the output named net1_output2 of the first model as the second input.
Review Comment:
```suggestion
# and uses the output named net1_output2 of the first model as the second input.
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -305,33 +485,29 @@ def test_pipeline():
pipe_config["input"]["data_b"].connect(pipe_config[mod2]["input"]["data_1"])
# The mod1 output[0] will be connected to a input named "data_0" of mod2.
- pipe_config[mod1]["output"][0].connect(pipe_config[mod2]["input"]["data_0"])
+ pipe_config[mod1]["output"][0].connect(pipe_config[mod2]["input"]["data_n_0"])
# The mod1 output[1] will be connected to a input named "data_0" of mod3.
- pipe_config[mod1]["output"][1].connect(pipe_config[mod3]["input"]["data_0"])
+ pipe_config[mod1]["output"][1].connect(pipe_config[mod3]["input"]["data_n_2"])
# The mod2 output[2] will be connected to a input named "data_1" of mod3.
- pipe_config[mod2]["output"][0].connect(pipe_config[mod3]["input"]["data_1"])
-
- # The mod1 output[2] will be connected to pipeline output[0].
- pipe_config[mod1]["output"][2].connect(pipe_config["output"]["0"])
+ pipe_config[mod2]["output"][0].connect(pipe_config[mod3]["input"]["data_n_1"])
- # The mod3 output[0] will be connected to pipeline output[1].
- pipe_config[mod3]["output"][0].connect(pipe_config["output"]["1"])
+ # The mod3 output[0] will be connected to pipeline output[0].
+ pipe_config[mod3]["output"][0].connect(pipe_config["output"]["0"])
# Print configueration (print(pipe_config)), the result looks like following.
Review Comment:
```suggestion
# Print configuration (print(pipe_config)), the result looks like following:
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
Review Comment:
```suggestion
# if the call has a free_var, recreate it.
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
Review Comment:
This is a string comparison, is this intended?.
split_operator_index sounds more like a integer.
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
+ # such constant may become free varaible due to the constant
+ # not exist, merge the previous constant with current subgraph
+ # to avoid such issue.
+ if constant_expr:
+ ann = merge_constant_expr(constant_expr, ann)
+ ann = run_opt_pass(ann, transform.ToGraphNormalForm())
+ mod = tvm.IRModule.from_expr(ann)
+ pipeline_mods.insert(0, mod)
+ # Return the last node of the current subgraph.
+ return tvm.relay.expr.Let(anf.var, value, body)
+ return tvm.relay.expr.Let(
+ anf.var,
+ value,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ )
+ else:
+ return anf
+
+ snode_dep = [{"nodes": {}, "ref_nodes": {}}]
+ pipeline_mods = []
+ operator_index_map = {}
+ # Used to tracking new input which caused by graph splitting.
+ new_input_idx = 0
+ constant_expr = None
+ subgraph_split_conf = split_conf.copy()
+ # Binding the parameters.
+ if params:
+ expr = build_module.bind_params_by_name(expr, params)
+ anf = run_opt_pass(expr, transform.ToANormalForm())
+ anf = run_opt_pass(anf, transform.InferType())
+ ann = _recursion(
+ anf,
+ pipeline_mods,
+ subgraph_split_conf,
+ constant_expr,
+ )
+ ann = run_opt_pass(ann.body, transform.ToGraphNormalForm())
+ mod = tvm.IRModule.from_expr(ann)
+ pipeline_mods.insert(0, mod)
+ return pipeline_mods
+
+
+def get_network():
+ # Get a list of modules representing subgraphs.
+ mods = []
+ dshape = (3, 3)
+ data = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ data21 = relay.var("data_1", relay.TensorType(dshape, "float32"))
+ data_net1_output_1 = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ data_net1_output_2 = relay.var("data_1", relay.TensorType(dshape, "float32"))
+ data_net2_output_1 = relay.var("data_0", relay.TensorType(dshape, "float32"))
+ mvalue1 = np.full((1), 1).astype("float32")
+ mvalue2 = np.full((1), 2).astype("float32")
+ mvalue3 = np.full((1), 3).astype("float32")
+ mv1 = relay.Constant(tvm.nd.array(mvalue1))
+ mv2 = relay.Constant(tvm.nd.array(mvalue2))
+ mv3 = relay.Constant(tvm.nd.array(mvalue3))
+ # There are three outputs in the first model.
+ net1_output1 = relay.add(data, mv1)
+ net1_output2 = relay.subtract(data, mv2)
+ net1_output3 = relay.concatenate((net1_output1, net1_output2), axis=0)
+ (net1_output3, _) = relay.split(net1_output3, indices_or_sections=2, axis=0)
+ net1_output3 = relay.add(net1_output3, mv2)
+ # The second model use output named net1_output1 of the first model as the first input,
Review Comment:
```suggestion
# The second model uses the output named net1_output1 of the first model as the first input,
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
Review Comment:
```suggestion
# merge constant express with an express
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
+ else:
+ new_args.append(var)
+ # if the call have a free_var, recreate it.
+ if need_update:
+ value = tvm.relay.expr.Call(
+ value.op, new_args, value.attrs, value.type_args, value.span
+ )
+ return value, snode_dep, new_input_idx
+
+ def merge_constant_expr(constant_expr, expr):
+ # merge constant express with a express
+ if not isinstance(constant_expr.body, tvm.relay.expr.Let):
+ return tvm.relay.expr.Let(constant_expr.var, constant_expr.value, expr)
+
+ return tvm.relay.expr.Let(
+ constant_expr.var, constant_expr.value, merge_constant_expr(constant_expr.body, expr)
+ )
+
+ def _recursion(anf, pipeline_mods, split_conf, constant_expr):
+ # Enumrate all operators of compute graph, then split the compute graph into a group of
+ # subgraph.
+ nonlocal operator_index_map
+ nonlocal new_input_idx
+ nonlocal snode_dep
+ cur_node_dep = snode_dep[len(snode_dep) - 1]
+ if isinstance(anf, tvm.relay.Function):
+ return tvm.relay.Function(
+ anf.params,
+ _recursion(anf.body, pipeline_mods, split_conf, constant_expr),
+ anf.ret_type,
+ anf.type_params,
+ anf.attrs,
+ )
+ if isinstance(anf, tvm.relay.expr.Let):
+ value = anf.value
+ # record the constant expr to make sure all sugraph can find correct constant.
+ if isinstance(value, tvm.relay.expr.Constant):
+ if not constant_expr:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, anf.var)
+ else:
+ constant_expr = tvm.relay.expr.Let(anf.var, value, constant_expr)
+ if isinstance(value, tvm.relay.expr.Call):
+ new_args = []
+ # build current var list
+ cur_node_dep["nodes"][anf.var] = 0
+ # Get the dependency information of the nodes.
+ value, snode_dep, new_input_idx = parse_dependency(value, snode_dep, new_input_idx)
+ if isinstance(value.op, tvm.ir.Op):
+ if value.op.name in operator_index_map:
+ operator_index_map[value.op.name] = operator_index_map[value.op.name] + 1
+ else:
+ operator_index_map[value.op.name] = 0
+ split_operator_name = split_conf[0]["op_name"] if split_conf else ""
+ split_operator_index = split_conf[0]["op_index"] if split_conf else ""
+ if (
+ split_conf
+ and split_operator_name in operator_index_map
+ and operator_index_map[split_operator_name] >= split_operator_index
+ ):
+ split_conf.pop(0)
+ snode_dep.append({"nodes": {}, "ref_nodes": {}})
+ ann = _recursion(
+ anf.body,
+ pipeline_mods,
+ split_conf,
+ constant_expr,
+ )
+ snode_dep.pop()
+ dep_vars = get_dep_var(snode_dep)
+ # When the nodes of current subgraph are the depedency node of other
+ # subgraph, we need to set them as the output of current subgraph.
+ body = relay.Tuple(dep_vars) if len(dep_vars) > 1 else anf.var
+ # when current subgraph use previous subgraph constant,
Review Comment:
```suggestion
# when the current subgraph uses a previous subgraph constant,
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
Review Comment:
```suggestion
dep["nodes"][var] += 1
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
+ if is_free_var:
+ need_update = True
+ new_args.append(relay.var(f"data_n_{new_input_idx}", var.checked_type))
+ new_input_idx = new_input_idx + 1
Review Comment:
```suggestion
new_input_idx += 1
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
Review Comment:
```suggestion
for deb in snode_dep[:-1]:
```
##########
tests/python/relay/test_pipeline_executor.py:
##########
@@ -22,12 +22,193 @@
import tvm
import tvm.testing
from tvm import relay
-from tvm.relay import transform
+from tvm.relay import transform, build_module
+from tvm.relay.testing import run_opt_pass
from tvm.contrib import graph_executor, pipeline_executor, pipeline_executor_build
from tvm._ffi import get_global_func
from tvm.contrib import cc as _cc
+"""Splitting graph into a list of subgraph"""
+
+
+def graph_split(expr, split_conf, params=None):
+ def get_dep_var(sub_var_dep):
+ return [var for var, _ in sub_var_dep[len(sub_var_dep) - 1]["ref_nodes"].items()]
+
+ def parse_dependency(value, snode_dep, new_input_idx):
+ new_args = []
+ need_update = False
+ for var in value.args:
+ is_free_var = False
+ for i in range(0, len(snode_dep) - 1):
+ dep = snode_dep[i]
+ if var in dep["nodes"]:
+ # Mark the previous subgraph node as a dependency.
+ dep["nodes"][var] = dep["nodes"][var] + 1
+ dep["ref_nodes"][var] = dep["nodes"][var]
+ # The var of this call is a free_var
+ is_free_var = True
+ # if the var of this call is free_var, recreate it and give it a fixed input name.
Review Comment:
```suggestion
# if the var of this call is a free_var, recreate it and give it a fixed input name.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [tvm] huajsj commented on pull request #11334: [Runtime][PipelineExecutor] Add graph manually splitting logic into the unit test.
Posted by GitBox <gi...@apache.org>.
huajsj commented on PR #11334:
URL: https://github.com/apache/tvm/pull/11334#issuecomment-1132297647
@masahi , please take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@tvm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org