Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions onnxscript/ir/passes/common/topological_sort.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
"""Pass for topologically sorting the graphs."""

from __future__ import annotations

__all__ = [
"TopologicalSortPass",
]


from onnxscript import ir


class TopologicalSortPass(ir.passes.InPlacePass):
"""Topologically sort graphs and functions in a model."""

def call(self, model: ir.Model) -> ir.passes.PassResult:
nodes = list(model.graph)
model.graph.sort()
new_nodes = list(model.graph)
for function in model.functions.values():
nodes.extend(function)
function.sort()
new_nodes.extend(function)

# Compare node orders to determine if any changes were made
modified = False
for node, new_node in zip(nodes, new_nodes):
if node is not new_node:
modified = True
break
return ir.passes.PassResult(model=model, modified=modified)
42 changes: 42 additions & 0 deletions onnxscript/ir/passes/common/topological_sort_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
"""Unit tests for the TopologicalSortPass."""

import unittest

from onnxscript import ir
from onnxscript.ir.passes.common import topological_sort


class TopologicalSortPassTest(unittest.TestCase):
def setUp(self):
self.node_a = ir.node("A", inputs=[], name="node_a")
self.node_b = ir.node("B", inputs=[self.node_a.outputs[0]], name="node_b")
self.node_c = ir.node("C", inputs=[self.node_b.outputs[0]], name="node_c")

def test_topological_sort_modified_true(self):
graph = ir.Graph(
inputs=self.node_a.inputs,
outputs=self.node_c.outputs,
nodes=[self.node_c, self.node_b, self.node_a], # Unsorted nodes
name="test_graph",
)
model = ir.Model(graph, ir_version=10)
pass_result = topological_sort.TopologicalSortPass()(model)
self.assertTrue(pass_result.modified)

def test_topological_sort_modified_false(self):
"""Test that modified is False when the input model is already sorted."""
sorted_graph = ir.Graph(
inputs=self.node_a.inputs,
outputs=self.node_c.outputs,
nodes=[self.node_a, self.node_b, self.node_c], # Sorted nodes
name="test_graph",
)
sorted_model = ir.Model(sorted_graph, ir_version=10)
pass_result = topological_sort.TopologicalSortPass()(sorted_model)
self.assertFalse(pass_result.modified)


if __name__ == "__main__":
unittest.main()
Loading