immutable_collections.py 2.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152
  1. from typing import Any, Dict, Tuple, List
  2. from ._compatibility import compatibility
  3. from torch.utils._pytree import Context, _register_pytree_node
  4. __all__ = ["immutable_list", "immutable_dict"]
  5. _help_mutation = """\
  6. If you are attempting to modify the kwargs or args of a torch.fx.Node object,
  7. instead create a new copy of it and assign the copy to the node:
  8. new_args = ... # copy and mutate args
  9. node.args = new_args
  10. """
  11. def _no_mutation(self, *args, **kwargs):
  12. raise NotImplementedError(f"'{type(self).__name__}' object does not support mutation. {_help_mutation}")
  13. def _create_immutable_container(base, mutable_functions):
  14. container = type('immutable_' + base.__name__, (base,), {})
  15. for attr in mutable_functions:
  16. setattr(container, attr, _no_mutation)
  17. return container
  18. immutable_list = _create_immutable_container(list,
  19. ['__delitem__', '__iadd__', '__imul__', '__setitem__', 'append',
  20. 'clear', 'extend', 'insert', 'pop', 'remove'])
  21. immutable_list.__reduce__ = lambda self: (immutable_list, (tuple(iter(self)),))
  22. compatibility(is_backward_compatible=True)(immutable_list)
  23. immutable_dict = _create_immutable_container(dict, ['__delitem__', '__setitem__', 'clear', 'pop', 'popitem', 'update'])
  24. immutable_dict.__reduce__ = lambda self: (immutable_dict, (iter(self.items()),))
  25. compatibility(is_backward_compatible=True)(immutable_dict)
  26. # Register immutable collections for PyTree operations
  27. def _immutable_dict_flatten(d: Dict[Any, Any]) -> Tuple[List[Any], Context]:
  28. return list(d.values()), list(d.keys())
  29. def _immutable_dict_unflatten(values: List[Any], context: Context) -> Dict[Any, Any]:
  30. return immutable_dict({key: value for key, value in zip(context, values)})
  31. def _immutable_list_flatten(d: List[Any]) -> Tuple[List[Any], Context]:
  32. return d, None
  33. def _immutable_list_unflatten(values: List[Any], context: Context) -> List[Any]:
  34. return immutable_list(values)
  35. _register_pytree_node(immutable_dict, _immutable_dict_flatten, _immutable_dict_unflatten)
  36. _register_pytree_node(immutable_list, _immutable_list_flatten, _immutable_list_unflatten)