docstr: The common event loop shared by all stages. This processses
  activations for the forward pass, and if `self.training` is true,
  processes gradients for the backward pass.
docstr_tok: ["The","common","event","loop","shared","by","all","stages",".","This","processses","activations","for","the","forward","pass",",","and","if","`","self.training","`","is","true",",","processes","gradients","for","the","backward","pass","."]
code: def event_loop_inner(
      self,
      expected_invocations: int,
      skip_trackers: List[SkipTrackerThroughPotals],
      activations: Activations,
      invocations: Invocations,
      count_per_order: Dict[int, int],
      *,
      already_received: int = 0,
      ignore_gradients: bool = False,
      event: Optional[Event] = None,
      tail: bool = False,
  ) -> None:
      """The common event loop shared by all stages. This processses
      activations for the forward pass, and if `self.training` is true,
      processes gradients for the backward pass."""
  
      num_activations = already_received
      if self.training and not ignore_gradients:
          num_gradients = 0
      else:
          num_gradients = expected_invocations
  
      while (
          num_activations < expected_invocations
          or num_gradients < expected_invocations
      ):
          if (
              num_activations == expected_invocations
              and num_gradients == 0
              and event is not None
          ):
              # We are ready to do the backward pass, but must wait for
              # PipeRPCWrapper to signal that it is safe to proceed, otherwise
              # deadlock
              event.wait()
  
          message = self.transport.recv_message_header(EVENT_LOOP_QUEUE)
          args: AsyncMessageBody = message.args
  
          invocation = invocations[args.order]
  
          # FIXME(tom) for combining pipeline with megatron, I currently don't
          # control the order of received activations or gradients, so it is
          # possible for a reused ColumnParallelLinear for example to receive
          # a different order of activations w.r.t. the sending stage, which
          # would result in incorrect values being used for the all_gather
          if args.message_type is AsyncMessageType.Activations:
              batch = self.get_batch_from_message(message)
  
              inv_count, last_order = self.run_invocations_on_batch(
                  batch,
                  invocations,
                  args.order,
                  skip_trackers,
                  activations,
              )
              count_per_order[last_order] = inv_count
              num_activations += inv_count
              if tail and invocations[last_order].dest is None:
                  self.prepare_tail_backward(
                      batch,
                      activations,
                      invocations,
                      count_per_order,
                      len(invocations) - inv_count,
                  )
  
              assert num_activations <= expected_invocations
  
          elif args.message_type is AsyncMessageType.Gradients:
              num_gradients += count_per_order[invocation.order]
              self.perform_backward_for_invocation(
                  self.transport, message, activations, invocation
              )
  
imports: from collections import OrderedDict
  from dataclasses import dataclass
  from enum import Enum, auto
  from threading import Event
  from typing import Dict, Iterable, List, Optional, Tuple
  import torch
  from torch import Tensor, nn
  from torch.autograd.profiler import record_function
  from torch.distributed import ProcessGroup
  from fairscale.nn.model_parallel import get_pipeline_parallel_ranks
  from .checkpoint import Checkpointing
  from .messages import Transport
  from .microbatch import Batch
  from .skip.tracker import SkipTrackerThroughPotals, use_skip_tracker
  from .types import EVENT_LOOP_QUEUE, PipeMessage, TensorOrTensors, Tensors
  from .worker import Task
label_desc: Use image node_id 11 for calling the AsyncEventLoop obj's underlying member method code with example usage: obj.event_loop_inner(expected_invocations, skip_trackers, activations, invocations, count_per_order) without return types