code: def _test(self, rampup_batch_size: Optional[List[int]]) -> None:
      for data_parallel_size in range(1, self.world_size + 1):
          expected_global_batch_size = self.GLOBAL_BATCH_SIZE
          expected_micro_batch_size = self.MICRO_BATCH_SIZE
          if rampup_batch_size:
              expected_global_batch_size = rampup_batch_size[0]
              num_consumed_samples = 0
              step_of_global_batch_size = rampup_batch_size[1]
              threshold = rampup_batch_size[2]
  
          if data_parallel_size > 1 and data_parallel_size % 2 != 0:
              continue
          if self.world_size % data_parallel_size != 0:
              continue
          msg = f"data_parallel_size: {data_parallel_size}"
          parallel_state.initialize_model_parallel(
              tensor_model_parallel_size_=self.world_size
              // data_parallel_size,
              pipeline_model_parallel_size_=1,
          )
          self.assertEqual(
              data_parallel_size,
              parallel_state.get_data_parallel_world_size(),
              msg=msg,
          )
  
          _reconfigure_microbatch_calculator(
              self.rank,
              rampup_batch_size,
              self.GLOBAL_BATCH_SIZE,
              self.MICRO_BATCH_SIZE,
              data_parallel_size,
          )
  
          self.assertEqual(
              get_micro_batch_size(), expected_micro_batch_size, msg=msg
          )
          self.assertEqual(
              get_num_microbatches(),
              expected_global_batch_size
              / expected_micro_batch_size
              / data_parallel_size,
              msg=msg,
          )
          current_global_batch_size = get_current_global_batch_size()
          self.assertEqual(
              current_global_batch_size,
              expected_global_batch_size,
              msg=msg,
          )
  
          # Make sure `global_batch_size` equals to the final global batch size after
          # certain number of updates.
          if rampup_batch_size:
              update_num_microbatches(current_global_batch_size)
              for i in range(100):
                  current_global_batch_size = (
                      get_current_global_batch_size()
                  )
                  update_num_microbatches(current_global_batch_size)
              current_global_batch_size = (
                  get_current_global_batch_size()
              )
              self.assertEqual(
                  get_current_global_batch_size(),
                  self.GLOBAL_BATCH_SIZE,
                  msg=msg,
              )
          parallel_state.destroy_model_parallel()
  
file_path: apex/tests/L0/run_transformer/test_microbatches.py
imports: import logging
  from typing import List, Optional
  from torch.testing._internal import common_utils
  from apex.transformer import parallel_state
  from apex.transformer.pipeline_parallel.utils import _reconfigure_microbatch_calculator, get_micro_batch_size, get_num_microbatches, get_current_global_batch_size, update_num_microbatches
  from apex.transformer.testing.distributed_test_base import NcclDistributedTestBase
  from apex.transformer.testing.distributed_test_base import UccDistributedTestBase
label_desc: Use image node_id 1 for calling the MicrobatchCalculatorTestBase obj's underlying member method code with example usage: obj._test(rampup_batch_size) without return types