c10d_logger.py 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. #!/usr/bin/env python3
  2. # mypy: allow-untyped-defs
  3. # Copyright (c) Facebook, Inc. and its affiliates.
  4. # All rights reserved.
  5. #
  6. # This source code is licensed under the BSD-style license found in the
  7. # LICENSE file in the root directory of this source tree.
  8. import functools
  9. import logging
  10. from typing import Any, Callable, TypeVar
  11. from typing_extensions import ParamSpec
  12. import torch
  13. import torch.distributed as dist
  14. from torch.distributed.logging_handlers import _log_handlers
  15. from torch.monitor import _WaitCounter
  16. __all__: list[str] = []
  17. _DEFAULT_DESTINATION = "default"
  18. def _get_or_create_logger(destination: str = _DEFAULT_DESTINATION) -> logging.Logger:
  19. logging_handler, log_handler_name = _get_logging_handler(destination)
  20. logger = logging.getLogger(f"c10d-{log_handler_name}")
  21. logger.setLevel(logging.DEBUG)
  22. formatter = logging.Formatter(
  23. "%(asctime)s %(filename)s:%(lineno)s %(levelname)s p:%(processName)s t:%(threadName)s: %(message)s"
  24. )
  25. logging_handler.setFormatter(formatter)
  26. logger.propagate = False
  27. logger.addHandler(logging_handler)
  28. return logger
  29. def _get_logging_handler(
  30. destination: str = _DEFAULT_DESTINATION,
  31. ) -> tuple[logging.Handler, str]:
  32. log_handler = _log_handlers[destination]
  33. log_handler_name = f"{type(log_handler).__name__}-{destination}"
  34. return (log_handler, log_handler_name)
  35. global _c10d_logger
  36. _c10d_logger = _get_or_create_logger()
  37. def _get_msg_dict(func_name, *args, **kwargs) -> dict[str, Any]:
  38. if dist.is_initialized():
  39. group = kwargs.get("group") or kwargs.get("process_group")
  40. msg_dict = {
  41. "func_name": f"{func_name}",
  42. "pg_name": f"{dist._get_process_group_name(kwargs.get('pg'))}", # type: ignore[arg-type]
  43. "backend": f"{dist.get_backend(group)}",
  44. "world_size": f"{dist.get_world_size()}",
  45. "group_size": f"{dist.get_world_size(group)}",
  46. "global_rank": f"{dist.get_rank()}",
  47. "local_rank": f"{dist.get_rank(group)}",
  48. }
  49. if msg_dict["backend"] == "nccl":
  50. nccl_version = torch.cuda.nccl.version()
  51. msg_dict["nccl_version"] = ".".join(str(v) for v in nccl_version)
  52. else:
  53. msg_dict = {
  54. "func_name": f"{func_name}",
  55. }
  56. return msg_dict
  57. _T = TypeVar("_T")
  58. _P = ParamSpec("_P")
  59. def _exception_logger(func: Callable[_P, _T]) -> Callable[_P, _T]:
  60. @functools.wraps(func)
  61. def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
  62. try:
  63. return func(*args, **kwargs)
  64. except Exception as error:
  65. msg_dict = _get_msg_dict(func.__name__, *args, **kwargs)
  66. msg_dict["error"] = f"{error}"
  67. _c10d_logger.debug(msg_dict)
  68. raise
  69. return wrapper
  70. def _time_logger(func: Callable[_P, _T]) -> Callable[_P, _T]:
  71. @functools.wraps(func)
  72. def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
  73. with _WaitCounter(f"pytorch.wait_counter.c10d.{func.__name__}").guard():
  74. func_return = func(*args, **kwargs)
  75. return func_return
  76. return wrapper