class_docstr: This class processes a single partition of the pipeline.
  Args:
      module_rref: RRef to the nn.Module for this partition. It should be on the local rpc worker.
      device: The device that holds the module.
      num_inputs: Numer of inputs to the module
      num_outputs: Number of outputs of the module. If the module output is not a tuple (and it is a
          single tensor), num_outputs should be None.
      rank: The rank of the partition
      chunks: Number of micor-batches in a mini-batch
      checkpoint_stop:: Checkpointing is done only for the first checkpoint_stop chunks of a mini-batch.
class_docstr_tok: ["This","class","processes","a","single","partition","of","the","pipeline",".","Args",":","module_rref",":","RRef","to","the","nn.Module","for","this","partition",".","It","should","be","on","the","local","rpc","worker",".","device",":","The","device","that","holds","the","module",".","num_inputs",":","Numer","of","inputs","to","the","module","num_outputs",":","Number","of","outputs","of","the","module",".","If","the","module","output","is","not","a","tuple","(","and","it","is","a","single","tensor",")",",","num_outputs","should","be","None",".","rank",":","The","rank","of","the","partition","chunks",":","Number","of","micor-batches","in","a","mini-batch","checkpoint_stop",":",":","Checkpointing","is","done","only","for","the","first","checkpoint_stop","chunks","of","a","mini-batch","."]
docstr: Processes a min-batch on this partition.
  If this is the last partition (pipeline_record has no consumer), concatenates results of processing
  all chunks and returns the result as the output of the model on the whole mini-batch.
docstr_tok: ["Processes","a","min-batch","on","this","partition",".","If","this","is","the","last","partition","(","pipeline_record","has","no","consumer",")",",","concatenates","results","of","processing","all","chunks","and","returns","the","result","as","the","output","of","the","model","on","the","whole","mini-batch","."]
code: def run_pipeline(
      self, pipeline_record_rref: rpc.RRef
  ) -> Optional[Tensor]:
      """Processes a min-batch on this partition.
      If this is the last partition (pipeline_record has no consumer), concatenates results of processing
      all chunks and returns the result as the output of the model on the whole mini-batch.
      """
      pipeline_record = pipeline_record_rref.local_value()
      self.run(pipeline_record)
  
      result: Optional[Tensor] = None
  
      if not pipeline_record.consumers:
          gather_result = microbatch.gather(pipeline_record.batches)
          assert len(gather_result) == 1
          result = gather_result[0]
          s0 = current_stream(result.device)
          if is_cuda(s0):
              # TODO. Investigate why this is needed and remove it if possible.
              as_cuda(s0).synchronize()
  
      # TODO: There seems to be a memory leak that is solved by following line.
      # Investigate why is it needed.
      del pipeline_record.batches
  
      return result
  
imports: from threading import Condition
  from types import TracebackType
  from typing import Dict, List, Optional, Tuple, Type, Union, cast
  import torch
  from torch import Tensor
  from torch.autograd.profiler import record_function
  from torch.distributed import rpc
  from fairscale.nn.pipe import microbatch
  from fairscale.nn.pipe.checkpoint import Checkpointing, TensorOrTensors
  from fairscale.nn.pipe.dependency import fork, join
  from fairscale.nn.pipe.microbatch import Batch
  from fairscale.nn.pipe.stream import as_cuda, current_stream, is_cuda, use_device, use_stream
  from fairscale.nn.pipe.worker import Task, create_workers
  from .data import DataConsumer
label_desc: Use image node_id 7 for calling the PartitionHandler obj's underlying member method code with example usage: obj.run_pipeline(pipeline_record_rref) and returns: result