class_docstr: Tasks store dictionaries and provide helpers for loading/iterating over
  Datasets, initializing the Model/Criterion and calculating the loss.
  
  Tasks have limited statefulness. In particular, state that needs to be
  saved to/loaded from checkpoints needs to be stored in the `self.state`
  :class:`StatefulContainer` object. For example::
  
      self.state.add_factory("dictionary", self.load_dictionary)
      print(self.state.dictionary)  # calls self.load_dictionary()
  
  This is necessary so that when loading checkpoints, we can properly
  recreate the task state after initializing the task instance.
class_docstr_tok: ["Tasks","store","dictionaries","and","provide","helpers","for","loading\/iterating","over","Datasets",",","initializing","the","Model\/Criterion","and","calculating","the","loss",".","Tasks","have","limited","statefulness",".","In","particular",",","state","that","needs","to","be","saved","to\/loaded","from","checkpoints","needs","to","be","stored","in","the","`","self.state","`",":","class",":","`","StatefulContainer","`","object",".","For","example",":",":","self.state.add_factory","(","``","dictionary","''",",","self.load_dictionary",")","print","(","self.state.dictionary",")","#","calls","self.load_dictionary","(",")","This","is","necessary","so","that","when","loading","checkpoints",",","we","can","properly","recreate","the","task","state","after","initializing","the","task","instance","."]
docstr: Aggregate logging outputs from data parallel training.
code: def reduce_metrics(self, logging_outputs, criterion):
      """Aggregate logging outputs from data parallel training."""
      # backward compatibility for tasks that override aggregate_logging_outputs
      base_func = FairseqTask.aggregate_logging_outputs
      self_func = getattr(self, "aggregate_logging_outputs").__func__
      if self_func is not base_func:
          utils.deprecation_warning(
              "Tasks should implement the reduce_metrics API. "
              "Falling back to deprecated aggregate_logging_outputs API."
          )
          agg_logging_outputs = self.aggregate_logging_outputs(
              logging_outputs, criterion
          )
          for k, v in agg_logging_outputs.items():
              metrics.log_scalar(k, v)
          return
  
      if not any("ntokens" in log for log in logging_outputs):
          warnings.warn(
              "ntokens not found in Criterion logging outputs, cannot log wpb or wps"
          )
      else:
          ntokens = sum(
              log.get("ntokens", 0) for log in logging_outputs
          )
          metrics.log_scalar("wpb", ntokens, priority=180, round=1)
          metrics.log_speed("wps", ntokens, priority=90, round=1)
  
      if not any("nsentences" in log for log in logging_outputs):
          warnings.warn(
              "nsentences not found in Criterion logging outputs, cannot log bsz"
          )
      else:
          nsentences = sum(
              log.get("nsentences", 0) for log in logging_outputs
          )
          metrics.log_scalar("bsz", nsentences, priority=190, round=1)
  
      criterion.__class__.reduce_metrics(logging_outputs)
  
imports: import logging
  import os
  import warnings
  from argparse import Namespace
  from typing import Any, Callable, Dict, List
  import torch
  from fairseq import search, tokenizer, utils
  from fairseq.logging import metrics
  from fairseq.data import Dictionary, FairseqDataset, data_utils, encoders, iterators
  from fairseq.dataclass import FairseqDataclass
  from fairseq.dataclass.utils import gen_parser_from_dataclass
  from fairseq.optim.amp_optimizer import AMPOptimizer
  from omegaconf import DictConfig
label_desc: Use image node_id 24 for calling the FairseqTask obj's underlying member method code with example usage: obj.reduce_metrics(logging_outputs, criterion) without return types