class_docstr: The main class template to implement a recommendation metric.
  This class contains the recommendation tasks information (RecTaskInfo) and
  the actual computation object (RecMetricComputation). RecMetric processes
  all the information related to RecTaskInfo and models, and passes the required
  signals to the computation object, allowing the implementation of
  RecMetricComputation to focus on the mathematical meaning.
  
  A new metric that inherits RecMetric must override the following attributes
  in its own `__init__()`: `_namespace` and `_metrics_computations`. No other
  methods should be overridden.
  
  Args:
      world_size (int): the number of trainers.
      my_rank (int): the rank of this trainer.
      batch_size (int): batch size used by this trainer.
      tasks (List[RecTaskInfo]): the information of the model tasks.
      compute_mode (RecComputeMode): the computation mode. See RecComputeMode.
      window_size (int): the window size for the window metric.
      fused_update_limit (int): the maximum number of updates to be fused.
      compute_on_all_ranks (bool): whether to compute metrics on all ranks. This
          is necessary if the non-leader rank wants to consume global metrics result.
      should_validate_update (bool): whether to check the inputs of `update()` and
          skip the update if the inputs are invalid. Invalid inputs include the case
          where all examples have 0 weights for a batch.
      process_group (Optional[ProcessGroup]): the process group used for the
          communication. Will use the default process group if not specified.
  
  Example::
  
      ne = NEMetric(
          world_size=4,
          my_rank=0,
          batch_size=128,
          tasks=DefaultTaskInfo,
      )
class_docstr_tok: ["The","main","class","template","to","implement","a","recommendation","metric",".","This","class","contains","the","recommendation","tasks","information","(","RecTaskInfo",")","and","the","actual","computation","object","(","RecMetricComputation",")",".","RecMetric","processes","all","the","information","related","to","RecTaskInfo","and","models",",","and","passes","the","required","signals","to","the","computation","object",",","allowing","the","implementation","of","RecMetricComputation","to","focus","on","the","mathematical","meaning",".","A","new","metric","that","inherits","RecMetric","must","override","the","following","attributes","in","its","own","`","__init__","(",")","`",":","`","_namespace","`","and","`","_metrics_computations","`",".","No","other","methods","should","be","overridden",".","Args",":","world_size","(","int",")",":","the","number","of","trainers",".","my_rank","(","int",")",":","the","rank","of","this","trainer",".","batch_size","(","int",")",":","batch","size","used","by","this","trainer",".","tasks","(","List","[","RecTaskInfo","]",")",":","the","information","of","the","model","tasks",".","compute_mode","(","RecComputeMode",")",":","the","computation","mode",".","See","RecComputeMode",".","window_size","(","int",")",":","the","window","size","for","the","window","metric",".","fused_update_limit","(","int",")",":","the","maximum","number","of","updates","to","be","fused",".","compute_on_all_ranks","(","bool",")",":","whether","to","compute","metrics","on","all","ranks",".","This","is","necessary","if","the","non-leader","rank","wants","to","consume","global","metrics","result",".","should_validate_update","(","bool",")",":","whether","to","check","the","inputs","of","`","update","(",")","`","and","skip","the","update","if","the","inputs","are","invalid",".","Invalid","inputs","include","the","case","where","all","examples","have","0","weights","for","a","batch",".","process_group","(","Optional","[","ProcessGroup","]",")",":","the","process","group","used","for","the","communication",".","Will","use","the","default","process","group","if","not","specified",".","Example",":",":","ne","=","NEMetric","(","world_size=4",",","my_rank=0",",","batch_size=128",",","tasks=DefaultTaskInfo",",",")"]
code: def __init__(
      self,
      world_size: int,
      my_rank: int,
      batch_size: int,
      tasks: List[RecTaskInfo],
      compute_mode: RecComputeMode = RecComputeMode.UNFUSED_TASKS_COMPUTATION,
      window_size: int = 100,
      fused_update_limit: int = 0,
      compute_on_all_ranks: bool = False,
      should_validate_update: bool = False,
      process_group: Optional[dist.ProcessGroup] = None,
      **kwargs: Dict[str, Any],
  ) -> None:
      torch._C._log_api_usage_once(
          f"torchrec.metrics.rec_metric.{self.__class__.__name__}"
      )
      # TODO(stellaya): consider to inherit from TorchMetrics.Metric or
      # TorchMetrics.MetricCollection.
      if (
          compute_mode == RecComputeMode.FUSED_TASKS_COMPUTATION
          and fused_update_limit > 0
      ):
          raise ValueError(
              "The fused tasks computation and the fused update cannot be set at the same time"
          )
      super().__init__()
      self._world_size = world_size
      self._my_rank = my_rank
      self._window_size = math.ceil(window_size / world_size)
      self._batch_size = batch_size
      self._metrics_computations = nn.ModuleList()
      self._tasks = tasks
      self._compute_mode = compute_mode
      self._fused_update_limit = fused_update_limit
      self._should_validate_update = should_validate_update
      self._default_weights = {}
      self._required_inputs = set()
      self._update_buffers = {
          self.PREDICTIONS: [],
          self.LABELS: [],
          self.WEIGHTS: [],
      }
      if self._window_size < self._batch_size:
          raise ValueError(
              f"Local window size must be larger than batch size. Got local window size {self._window_size} and batch size {self._batch_size}."
          )
  
      if compute_mode == RecComputeMode.FUSED_TASKS_COMPUTATION:
          task_per_metric = len(self._tasks)
          self._tasks_iter = self._fused_tasks_iter
      else:
          task_per_metric = 1
          self._tasks_iter = self._unfused_tasks_iter
  
      for task_config in (
          [self._tasks]
          if compute_mode == RecComputeMode.FUSED_TASKS_COMPUTATION
          else self._tasks
      ):
          # pyre-ignore
          kwargs["fused_update_limit"] = fused_update_limit
          # This Pyre error seems to be Pyre's bug as it can be inferred by mypy
          # according to https://github.com/python/mypy/issues/3048.
          # pyre-fixme[45]: Cannot instantiate abstract class `RecMetricCoputation`.
          metric_computation = self._computation_class(
              my_rank,
              batch_size,
              task_per_metric,
              self._window_size,
              compute_on_all_ranks,
              self._should_validate_update,
              process_group,
              **{**kwargs, **self._get_task_kwargs(task_config)},
          )
          required_inputs = self._get_task_required_inputs(task_config)
  
          self._metrics_computations.append(metric_computation)
          self._required_inputs.update(required_inputs)
  
imports: import abc
  import itertools
  import math
  from collections import defaultdict, deque
  from dataclasses import dataclass
  from enum import Enum
  from typing import Any, Callable, cast, Deque, Dict, Iterator, List, Mapping, Optional, Sequence, Set, Tuple, Type, TypeVar, Union
  import torch
  import torch.distributed
  import torch.nn
  from torch.profiler import record_function
  from torchmetrics import Metric
  from torchrec.distributed.types import get_tensor_size_bytes
  from torchrec.metrics.metrics_config import RecComputeMode, RecTaskInfo
  from torchrec.metrics.metrics_namespace import compose_metric_key, MetricNameBase, MetricNamespaceBase, MetricPrefix
label_desc: Use image node_id 1 to create a new RecMetric object from inherited base classes: nn, abc with example: obj = RecMetric(world_size, my_rank, batch_size, tasks, compute_mode, window_size, fused_update_limit, compute_on_all_ranks, should_validate_update, process_group)