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 _exec_forward_pass(self, buffer_id):
      self.tput_timer.start()
      self.mem_status("BEFORE FWD", reset_max=True)
  
      if isinstance(self.pipe_buffers["inputs"][buffer_id], tuple):
          inputs = tuple(
              t.clone() for t in self.pipe_buffers["inputs"][buffer_id]
          )
      else:
          inputs = self.pipe_buffers["inputs"][buffer_id].clone()
  
      # collect the partitioned input from the previous stage
      if self.is_pipe_partitioned and not self.is_first_stage():
          if self.pipe_partition_input_meta_cache is None:
              self.pipe_partition_input_meta_cache = inputs[0].to("cpu")
          part_input = PartitionedTensor.from_meta(
              meta=self.pipe_partition_input_meta_cache,
              local_part=inputs[1],
              group=self.grid.get_slice_parallel_group(),
          )
  
          inputs = (part_input.full(), *inputs[2:])
          inputs[0].requires_grad = True
          # skip mask
          # inputs[1].requires_grad = True
          part_input = None
          inputs = inputs[0] if len(inputs) == 1 else inputs
          self.pipe_buffers["inputs"][buffer_id] = inputs
  
      # inputs has no gradient because it is from a cloned tensor
      outputs = super().forward(inputs)
  
      # Reset activation checkpointing buffers.
      # Need to call this between evaluation iterations
      if not self.module.training:
          ds_checkpointing.reset()
  
      # Partition the outputs if we are not the last stage
      if self.is_pipe_partitioned and not self.is_last_stage():
          if isinstance(outputs, tuple):
              first_output = outputs[0]
              # TODO: Improve pipe partitioning to pass multiple tensors that require grads
              assert all(
                  [
                      torch.is_tensor(elt)
                      and elt.requires_grad is False
                      for elt in outputs[1:]
                  ]
              )
              outputs_tail = outputs[1:]
          elif torch.is_tensor(outputs):
              first_output = outputs
              outputs_tail = []
          else:
              raise ValueError(
                  "expecting a tensor or a tuple of tensors"
              )
          part = PartitionedTensor(
              tensor=first_output,
              group=self.grid.get_slice_parallel_group(),
          )
          # Clear the large output data, but save the computation graph
          first_output.data = torch.zeros(1)
          self.pipe_buffers["output_tensors"][buffer_id] = first_output
          # Inject the partitioned tensor into the output before sending
          outputs = (part.to_meta(), part.data(), *outputs_tail)
          part = None
  
      self.pipe_buffers["outputs"][buffer_id] = outputs
  
      # Optionally compute loss on the last device
      if self.is_last_stage():
          if self._compute_loss and self.module.loss_fn is not None:
              labels = self.pipe_buffers["labels"][buffer_id]
              self.loss = self.module.loss_fn(outputs, labels)
          else:
              # Some models just return loss from forward()
              self.loss = outputs
          if self.eval_return_logits:
              self.outputs = outputs
          if isinstance(self.loss, torch.Tensor):
              self.fwd_outputs.append(self.loss.detach())
  
              if self.total_loss is None:
                  self.total_loss = torch.zeros_like(self.loss)
              self.total_loss += self.loss.detach()
          else:
              self.fwd_outputs.append([l.detach() for l in self.loss])
  
              if self.total_loss is None:
                  self.total_loss = [
                      torch.zeros_like(l) for l in self.loss
                  ]
              for idx, l in enumerate(self.loss):
                  self.total_loss[idx] += l.detach()
  
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 24 for calling the PipelineEngine obj's underlying member method code with example usage: obj._exec_forward_pass(buffer_id) without return types