class_docstr: The async pipeline parallelism for Pipe.
class_docstr_tok: ["The","async","pipeline","parallelism","for","Pipe","."]
docstr: Runs pipeline parallelism.
  
  It modifies the given batches in place.
docstr_tok: ["Runs","pipeline","parallelism",".","It","modifies","the","given","batches","in","place","."]
code: def run(
      self, training: bool, batches: List[Batch], event: Optional[Event]
  ) -> None:
      """Runs pipeline parallelism.
  
      It modifies the given batches in place.
  
      """
      self.training = training
  
      skip_trackers = [
          SkipTrackerThroughPotals(self.skip_layout, i)
          for i in range(len(batches))
      ]
  
      rank = self.group.rank()
      event_loop = AsyncEventLoop(
          self.partitions,
          self.group,
          self.transport,
          self.training,
          self.checkpoint_stop,
      )
      if rank == 0 and not self.final_stage:
          logging.debug(
              f"{torch.distributed.get_rank()}: entered event head"
          )
          event_loop.event_loop_head(batches, skip_trackers, event)
          logging.debug(
              f"{torch.distributed.get_rank()}: exited event head"
          )
      elif self.final_stage:
          logging.debug(
              f"{torch.distributed.get_rank()}: entered event tail"
          )
          event_loop.event_loop_tail(batches, skip_trackers)
          logging.debug(
              f"{torch.distributed.get_rank()}: exited event tail"
          )
      else:
          logging.debug(
              f"{torch.distributed.get_rank()}: entered event loop"
          )
          event_loop.event_loop(len(batches), skip_trackers)
          logging.debug(
              f"{torch.distributed.get_rank()}: exited event loop"
          )
  
file_path: fairscale/fairscale/nn/pipe/async_pipeline.py
imports: import logging
  import os
  from threading import Event
  from typing import Dict, List, Optional, Union
  import torch
  from .async_schedule import AsyncEventLoop, ModuleWrapper
  from .messages import MakeTransport
  from .microbatch import Batch
  from .skip.layout import SkipLayout
  from .skip.tracker import SkipTrackerThroughPotals
label_desc: Use image node_id 3 for calling the AsyncPipeline obj's underlying member method code with example usage: obj.run(training, batches, event) without return types