class_docstr: The internal computation class template.
  A metric implementation should overwrite `update()` and `compute()`. These two
  APIs focus on the actual mathematical meaning of the metric, without detailed
  knowledge of model output and task information.
  
  Args:
      my_rank (int): the rank of this trainer.
      batch_size (int): batch size used by this trainer.
      n_tasks (int): the number tasks this communication object will have to compute.
      window_size (int): the window size for the window metric.
      compute_on_all_ranks (bool): whether to compute metrics on all ranks. This is
          necessary if the non-leader rank wants to consume the metrics results.
      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.
class_docstr_tok: ["The","internal","computation","class","template",".","A","metric","implementation","should","overwrite","`","update","(",")","`","and","`","compute","(",")","`",".","These","two","APIs","focus","on","the","actual","mathematical","meaning","of","the","metric",",","without","detailed","knowledge","of","model","output","and","task","information",".","Args",":","my_rank","(","int",")",":","the","rank","of","this","trainer",".","batch_size","(","int",")",":","batch","size","used","by","this","trainer",".","n_tasks","(","int",")",":","the","number","tasks","this","communication","object","will","have","to","compute",".","window_size","(","int",")",":","the","window","size","for","the","window","metric",".","compute_on_all_ranks","(","bool",")",":","whether","to","compute","metrics","on","all","ranks",".","This","is","necessary","if","the","non-leader","rank","wants","to","consume","the","metrics","results",".","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","."]
code: def __init__(
      self,
      my_rank: int,
      batch_size: int,
      n_tasks: int,
      window_size: int,
      compute_on_all_ranks: bool = False,
      should_validate_update: bool = False,
      process_group: Optional[dist.ProcessGroup] = None,
      fused_update_limit: int = 0,
      allow_missing_label_with_zero_weight: bool = False,
      *args: Any,
      **kwargs: Any,
  ) -> None:
      super().__init__(process_group=process_group, *args, **kwargs)
  
      self._my_rank = my_rank
      self._n_tasks = n_tasks
      self._batch_size = batch_size
      self._window_size = window_size
      self._compute_on_all_ranks = compute_on_all_ranks
      self._should_validate_update = should_validate_update
      if self._window_size > 0:
          self._batch_window_buffers = {}
      else:
          self._batch_window_buffers = None
      if self._should_validate_update:
          self._add_state(
              "has_valid_update",
              torch.zeros(self._n_tasks, dtype=torch.uint8),
              add_window_state=False,
              dist_reduce_fx=lambda x: torch.any(x, dim=0).byte(),
              persistent=True,
          )
  
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 RecMetricComputation object from inherited base classes: Metric, abc with example: obj = RecMetricComputation(my_rank, batch_size, n_tasks, window_size, compute_on_all_ranks, should_validate_update, process_group, fused_update_limit, allow_missing_label_with_zero_weight)