(applied_map:dict[UOp, UOp], name:str, walk:bool=False)
| 21 | |
| 22 | all_tensors: dict[weakref.ref[Tensor], None] = {} |
| 23 | def _apply_map_to_tensors(applied_map:dict[UOp, UOp], name:str, walk:bool=False) -> None: |
| 24 | with cpu_profile(TracingKey(name), "TINY"): |
| 25 | # get tensors in scope |
| 26 | in_scope: dict[UOp, bool] = {} |
| 27 | def visitor(node: UOp) -> bool: return True if node in applied_map else any(in_scope.get(s, False) for s in node.src) |
| 28 | scope_tensors: list[Tensor] = [t for tref in list(all_tensors) if (t:=tref()) is not None and t.uop.topovisit(visitor, in_scope)] |
| 29 | |
| 30 | # get all Tensors and apply the map |
| 31 | sink = UOp.sink(*[t.uop for t in scope_tensors]) |
| 32 | new_sink = sink.substitute(applied_map, name=f"substitute {name}", walk=walk) |
| 33 | |
| 34 | # set the relevant uop to the realized UOps |
| 35 | for t,s,ns in zip(scope_tensors, sink.src, new_sink.src): |
| 36 | if s is ns: continue |
| 37 | t.uop = ns |
| 38 | |
| 39 | # **** Tensor helper functions **** |
| 40 |
no test coverage detected
searching dependent graphs…