(id2name, awaits, task_stacks)
| 60 | |
| 61 | |
| 62 | def _build_tree(id2name, awaits, task_stacks): |
| 63 | id2label = {(NodeType.TASK, tid): name for tid, name in id2name.items()} |
| 64 | children = defaultdict(list) |
| 65 | cor_nodes = defaultdict(dict) # Maps parent -> {frame_name: node_key} |
| 66 | next_cor_id = count(1) |
| 67 | |
| 68 | def get_or_create_cor_node(parent, frame): |
| 69 | """Get existing coroutine node or create new one under parent""" |
| 70 | if frame in cor_nodes[parent]: |
| 71 | return cor_nodes[parent][frame] |
| 72 | |
| 73 | node_key = (NodeType.COROUTINE, f"c{next(next_cor_id)}") |
| 74 | id2label[node_key] = frame |
| 75 | children[parent].append(node_key) |
| 76 | cor_nodes[parent][frame] = node_key |
| 77 | return node_key |
| 78 | |
| 79 | # Build task dependency tree with coroutine frames |
| 80 | for parent_id, stack, child_id in awaits: |
| 81 | cur = (NodeType.TASK, parent_id) |
| 82 | for frame in reversed(stack): |
| 83 | cur = get_or_create_cor_node(cur, frame) |
| 84 | |
| 85 | child_key = (NodeType.TASK, child_id) |
| 86 | if child_key not in children[cur]: |
| 87 | children[cur].append(child_key) |
| 88 | |
| 89 | # Add coroutine stacks for leaf tasks |
| 90 | awaiting_tasks = {parent_id for parent_id, _, _ in awaits} |
| 91 | for task_id in id2name: |
| 92 | if task_id not in awaiting_tasks and task_id in task_stacks: |
| 93 | cur = (NodeType.TASK, task_id) |
| 94 | for frame in reversed(task_stacks[task_id]): |
| 95 | cur = get_or_create_cor_node(cur, frame) |
| 96 | |
| 97 | return id2label, children |
| 98 | |
| 99 | |
| 100 | def _roots(id2label, children): |
no test coverage detected
searching dependent graphs…