class_docstr: A training engine hybrid pipeline, data, and model parallel training.
  
  This engine is created by ``deepspeed.initialize()`` when a :class:`PipelineModule`
  is provided.
class_docstr_tok: ["A","training","engine","hybrid","pipeline",",","data",",","and","model","parallel","training",".","This","engine","is","created","by","``","deepspeed.initialize","(",")","``","when","a",":","class",":","`","PipelineModule","`","is","provided","."]
code: def _aggregate_total_loss(self):
      # Scale loss, average among DP ranks, and bcast loss to the rest of my DP group
      if self.is_last_stage():
          loss = self._scale_loss_by_gas(self.total_loss)
          self.dp_group_loss = loss.clone().detach()
  
          ## Average loss across all data-parallel groups
          agg_loss = self.dp_group_loss.clone().detach()
          # print(f'RANK={self.global_rank} bcast SENDER src={self.global_rank} group={self.grid.pp_group}', flush=True)
          if self.is_data_parallel:
              dist.all_reduce(
                  agg_loss, group=self.mpu.get_data_parallel_group()
              )
              agg_loss /= self.dp_world_size
  
          assert self.global_rank in self.grid.pp_group
          losses = torch.stack([self.dp_group_loss, agg_loss]).float()
          if self.is_pipe_parallel:
              dist.broadcast(
                  tensor=losses,
                  src=self.global_rank,
                  group=self.mpu.get_pipe_parallel_group(),
              )
      else:
          # Get loss from last stage
          src_rank = self.grid.stage_to_global(self.num_stages - 1)
          assert src_rank in self.grid.pp_group
          losses = torch.Tensor([0.0, 0.0]).to(self.device)
          dist.broadcast(
              tensor=losses,
              src=src_rank,
              group=self.grid.get_pipe_parallel_group(),
          )
          self.dp_group_loss = losses[0].clone().detach()
          agg_loss = losses[1].clone().detach()
  
      return agg_loss
  
imports: from types import MethodType
  import torch
  from deepspeed import comm
  from deepspeed.utils import logger
  from deepspeed.utils.timer import ThroughputTimer
  from deepspeed.accelerator import get_accelerator
  from deepspeed.runtime.bf16_optimizer import BF16_Optimizer
  from ..engine import DeepSpeedEngine, MEMORY_OPT_ALLREDUCE_SIZE
  from deepspeed.utils.timer import FORWARD_MICRO_TIMER, FORWARD_GLOBAL_TIMER, BACKWARD_MICRO_TIMER, BACKWARD_GLOBAL_TIMER, BACKWARD_INNER_MICRO_TIMER, BACKWARD_INNER_GLOBAL_TIMER, BACKWARD_REDUCE_MICRO_TIMER, BACKWARD_REDUCE_GLOBAL_TIMER, STEP_MICRO_TIMER, STEP_GLOBAL_TIMER
  from ..utils import PartitionedTensor
  from ..dataloader import RepeatingLoader
  from ..zero.config import ZeroStageEnum
  from ..activation_checkpointing import checkpointing
  from .module import PipelineModule, PipelineError
  from .None import p2p
  from .None import schedule
label_desc: Use image node_id 16 for calling the PipelineEngine obj's underlying member method code with example usage: obj._aggregate_total_loss() and returns: agg_loss