Keep separate per-pipeline operator counters. Error out when "stealing" subgraphs from other pipelines results in duplicate names. #5506
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
mzient
commented
Jun 6, 2024
Comment on lines
-2104
to
-2134
| def get_op_outputs_num(): | ||
| # BSF traverse the graph first to learn, for each reachable operator in the graph, | ||
| # how many data-nodes/edges the operator contributes to | ||
| # (i.e. the number of outputs of the operator instance) | ||
| op_outputs_num = {} | ||
| edges = deque(output_nodes) | ||
| while edges: | ||
| current_edge = edges.popleft() | ||
| source_op = get_source_op(current_edge) | ||
| if source_op.id in op_outputs_num: | ||
| op_outputs_num[source_op.id] += 1 | ||
| else: | ||
| op_outputs_num[source_op.id] = 1 | ||
| source_op.check_args() | ||
| edges.extend(get_op_input_edges(source_op)) | ||
| return op_outputs_num | ||
|
|
||
| visited = set() | ||
| ops = [] | ||
| edges = deque(output_nodes) | ||
| op_total_outputs_num = get_op_outputs_num() | ||
| op_visited_outputs_num = {op_id: 0 for op_id in op_total_outputs_num} | ||
| while edges: | ||
| current_edge = edges.popleft() | ||
| source_op = get_source_op(current_edge) | ||
| op_visited_outputs_num[source_op.id] += 1 | ||
| # Actually visit the operator only when all the nodes it contributes to | ||
| # were already processed | ||
| if op_visited_outputs_num[source_op.id] == op_total_outputs_num[source_op.id]: | ||
| ops.append(source_op) | ||
| edges.extend(get_op_input_edges(source_op)) | ||
| ops.reverse() |
There was a problem hiding this comment.
This was unnecessarily complex. DFS is much better suited for discovering topology.
mzient
commented
Jun 6, 2024
| @@ -57,7 +57,8 @@ | |||
| class _OpCounter(object): | |||
| # pylint: disable=too-few-public-methods | |||
| _lock = threading.Lock() | |||
| _op_count = count(0) | |||
| # start from something large to avoid confusion with (more common) per-pipeline numbering | |||
There was a problem hiding this comment.
Note for reviewers: "avoid confusion" means only confusion while debugging. The renaming can deal with duplicate names.
|
CI MESSAGE: [15624008]: BUILD STARTED |
szkarpinski
reviewed
Jun 6, 2024
| @@ -57,7 +57,8 @@ | |||
| class _OpCounter(object): | |||
| # pylint: disable=too-few-public-methods | |||
| _lock = threading.Lock() | |||
| _op_count = count(0) | |||
| # start from something large to avoid confusion with (more common) per-pipeline numbering | |||
There was a problem hiding this comment.
Suggested change
| # start from something large to avoid confusion with (more common) per-pipeline numbering | |
| # start from something large to avoid confusion while debugging with (more common) per-pipeline numbering |
|
CI MESSAGE: [15624354]: BUILD STARTED |
|
CI MESSAGE: [15624997]: BUILD STARTED |
|
CI MESSAGE: [15624997]: BUILD FAILED |
|
CI MESSAGE: [15633718]: BUILD STARTED |
|
CI MESSAGE: [15633718]: BUILD FAILED |
|
CI MESSAGE: [15654084]: BUILD STARTED |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
CI MESSAGE: [15654084]: BUILD FAILED |
|
CI MESSAGE: [15654586]: BUILD STARTED |
mzient
changed the title
mzient
changed the title
mzient
force-pushed
the
per_pipeline_name_generation
branch
from
6ac2b99 to
a291562
Compare
|
CI MESSAGE: [15709123]: BUILD FAILED |
…es from other pipelines (including None). Signed-off-by: Michal Zientkiewicz <[email protected]>
Signed-off-by: Michal Zientkiewicz <[email protected]>
Signed-off-by: Michal Zientkiewicz <[email protected]>
Signed-off-by: Michal Zientkiewicz <[email protected]>
- handle renaming of duplicate nodes - handle renaming of pipeline inputs that are not direct outputs of any operator - improve error message Signed-off-by: Michal Zientkiewicz <[email protected]>
Signed-off-by: Michal Zientkiewicz <[email protected]>
Signed-off-by: Michal Zientkiewicz <[email protected]>
Signed-off-by: Michal Zientkiewicz <[email protected]>
…ends. Signed-off-by: Michal Zientkiewicz <[email protected]>
mzient
force-pushed
the
per_pipeline_name_generation
branch
from
a291562 to
999c1cc
Compare
|
CI MESSAGE: [15734267]: BUILD STARTED |
szkarpinski
reviewed
Jun 11, 2024
szkarpinski
reviewed
Jun 11, 2024
szkarpinski
approved these changes
Jun 11, 2024
|
CI MESSAGE: [15734267]: BUILD PASSED |
banasraf
approved these changes
Jun 11, 2024
18 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Category:
New feature (non-breaking change which adds functionality)
Refactoring (Redesign of existing code that doesn't affect functionality)
Description:
Prior to this change constructing exactly the same pipeline (e.g. by calling a function decorated with
@pipeline_def) multiple times produced pipelines with different operator instance names and differently named operator instances and DataNodes. This PR changes that so that pipelines with the same structure have the same node names.This is achieved by:
Pipelines that are defined without a "current" pipeline have distinct operator instance names.
Additionally, there were some problems with operator discovery. I rewrote it to a much simpler DFS.
Additional information:
Affected modules and functionalities:
Key points relevant for the review:
Tests:
pipeline_test.py: test_dangling_subgraph
Checklist
Documentation
DALI team only
Requirements
REQ IDs: N/A
JIRA TASK: N/A