| 403 | |
| 404 | |
| 405 | class DispatchTracer(fx.Tracer): |
| 406 | def __init__(self) -> None: |
| 407 | super().__init__() |
| 408 | self.root: torch.nn.Module = torch.nn.Module() |
| 409 | self.tensor_attrs: Dict[torch.Tensor, str] = {} |
| 410 | self.submodules: Dict[fx.GraphModule, str] = {} |
| 411 | |
| 412 | def call_module( |
| 413 | self, |
| 414 | m: torch.nn.Module, |
| 415 | forward: Callable[..., Value], |
| 416 | args: Tuple[Value, ...], |
| 417 | kwargs: Dict[str, Value], |
| 418 | ) -> Value: |
| 419 | return forward(*args, **kwargs) |
| 420 | |
| 421 | def _module_getattr( |
| 422 | self, attr: str, attr_val: Value, parameter_proxy_cache: Dict[str, torch.Tensor] |
| 423 | ) -> Value: |
| 424 | if isinstance(attr_val, torch.nn.Parameter): |
| 425 | for n, p in self.root.named_parameters(): |
| 426 | if attr_val is p: |
| 427 | if n not in parameter_proxy_cache: |
| 428 | proxy = self.create_proxy("get_attr", n, (), {}) |
| 429 | parameter_proxy_cache[n] = PythonTensor(attr_val, proxy) |
| 430 | return parameter_proxy_cache[n] |
| 431 | return attr_val |
| 432 | return attr_val |
| 433 | |
| 434 | def create_arg(self, a: Value) -> torch.fx.Node: # noqa: C901 |
| 435 | if isinstance(a, torch.nn.Parameter): |
| 436 | for n, p in self.root.named_parameters(): |
| 437 | if a is p: |
| 438 | return self.create_node("get_attr", n, (), {}) |
| 439 | qualname: Optional[str] = None |
| 440 | |
| 441 | if not qualname: |
| 442 | i = 0 |
| 443 | while True: |
| 444 | qualname = f"_param_constant{i}" |
| 445 | if not hasattr(self.root, qualname): |
| 446 | break |
| 447 | i += 1 |
| 448 | setattr(self.root, qualname, a) |
| 449 | |
| 450 | return self.create_node("get_attr", qualname, (), {}) |
| 451 | |
| 452 | if isinstance(a, torch.Tensor): |
| 453 | qualname: Optional[str] = self.tensor_attrs.get(a) |
| 454 | |
| 455 | if not qualname: |
| 456 | i = 0 |
| 457 | while True: |
| 458 | qualname = f"_tensor_constant{i}" |
| 459 | if not hasattr(self.root, qualname): |
| 460 | break |
| 461 | i += 1 |
| 462 | self.tensor_attrs[a] = qualname |
no outgoing calls
no test coverage detected