code: def _generate_fancy_data_labels(self, sequence_len, batch_size):
      temps = []
      for i in range(batch_size):
          if self.inds is None or self.data_idx >= len(self.inds):
              # hack as use of RNG will fall out of sync due to pipelines being different
              torch.manual_seed(self.MANUAL_SEED)
              self.inds = torch.randperm(
                  self.effective_length, device="cuda"
              )
              self.masks = (
                  torch.rand(
                      len(self.inds) // batch_size + 1,
                      batch_size,
                      sequence_len,
                      device="cuda",
                  )
                  >= self.MASK_PROB
              ).long()
              self.MANUAL_SEED += 1
              self.data_idx = 0
              if self.rank == 0:
                  print("new epoch", len(self.inds))
                  print("my start", self.inds[0:5])
                  print("masks_checksum:", torch.sum(self.masks))
          if self.EASY_MODE:
              data_idx_ = self.data_idx % self.EASY_MODE_SIZ
          else:
              data_idx_ = self.data_idx
          offset = self.inds[data_idx_]  # * SEQUENCE_LEN
          self.data_idx += 1
  
          curr = (
              self.fancy_data[offset : offset + sequence_len]
              .clone()
              .detach()
          )
          temps.append(curr)
      temp = torch.stack(temps, dim=0).cuda()
      mask = self.masks[self.data_idx // batch_size]
      mask_not = torch.logical_not(mask).long()
      data = mask * temp + mask_not * 124
      label = temp
      if parallel_state.get_tensor_model_parallel_rank() == 0:
          data_dict = {
              "text": data,
              "label": label,
              "mask_not": mask_not,
          }
      else:
          data_dict = None
      keys = ["text", "label", "mask_not"]
      broadcasted_data = tensor_parallel.broadcast_data(
          keys, data_dict, torch.long
      )
      return (
          broadcasted_data["text"].long(),
          broadcasted_data["label"].long(),
          broadcasted_data["mask_not"],
      )
  
imports: import torch
  import unittest
  from apex.transformer.testing import global_vars
  from apex.transformer.testing.standalone_bert import bert_model_provider
  from apex.transformer.pipeline_parallel.schedules.common import _get_params_for_weight_decay_optimization, build_model
  from apex.transformer.pipeline_parallel.schedules import get_forward_backward_func
  from apex.transformer.pipeline_parallel.utils import average_losses_across_data_parallel_group, unwrap_model, setup_microbatch_calculator
  from apex.transformer.log_util import set_logging_level
  from apex.transformer import tensor_parallel, parallel_state
  from apex.transformer.enums import ModelType
  from apex.transformer._ucc_util import HAS_UCC
  from apex.transformer.testing.distributed_test_base import UccDistributedTestBase, NcclDistributedTestBase
  import logging
  from torch.testing._internal import common_utils
label_desc: Use image node_id 2 for calling the BertTestBase obj's underlying member method code with example usage: obj._generate_fancy_data_labels(sequence_len, batch_size) and returns: broadcasted_data,  broadcasted_data,  broadcasted_data