code: def test_4(self):
      data_length = 100
      torch_normal_data = TorchNormalDataset(num_of_data=data_length)
      random_sampler_1 = partial(
          RandomSampler, dataset=torch_normal_data, shuffle=True
      )
  
      world_size_1 = 2
      forward_steps = 10
  
      for pad in {True, False}:
          all_rank_state = {}
          all_rank_first_left_data = {}
          all_rank_second_epoch = {}
          for rank in range(world_size_1):
              each_rank_iterable = random_sampler_1()
              each_rank_iterable.set_distributed(
                  num_replicas=world_size_1, rank=rank, pad=pad
              )
              iterator = iter(each_rank_iterable)
              # 第一轮
              each_rank_iterable.set_epoch(0)
              first_epoch = []
              for _ in range(forward_steps):
                  first_epoch.append(next(iterator))
  
              # 先提前保存断点重训的结果;
              all_rank_state[rank] = each_rank_iterable.state_dict()
  
              # 保存第一个 epoch 的之后的结果，用于查看断点重训是否正确；
              first_left_data = []
              while True:
                  try:
                      first_left_data.append(next(iterator))
                  except StopIteration:
                      break
              all_rank_first_left_data[rank] = first_left_data
              # 第二轮
              each_rank_iterable.set_epoch(1)
              iterator = iter(each_rank_iterable)
              second_epoch = []
              for _ in range(forward_steps):
                  second_epoch.append(next(iterator))
              all_rank_second_epoch[rank] = second_epoch
              assert first_epoch != second_epoch
  
          # 重新加载第一轮的状态，查看断点重训是否正确；
          random_sampler_2 = partial(
              RandomSampler, dataset=torch_normal_data, shuffle=True
          )
          for rank in range(world_size_1):
              each_rank_iterable = random_sampler_2()
              each_rank_iterable.set_distributed(
                  num_replicas=world_size_1, rank=rank, pad=pad
              )
              each_rank_iterable.load_state_dict(all_rank_state[rank])
              each_rank_iterable.set_epoch(0)
              iterator = iter(each_rank_iterable)
              re_first_epoch = []
              while True:
                  try:
                      re_first_epoch.append(next(iterator))
                  except StopIteration:
                      break
              assert re_first_epoch == all_rank_first_left_data[rank]
  
              # 查看第二轮的结果是否也是和第一次的第二轮完全一致；
              each_rank_iterable.set_epoch(1)
              iterator = iter(each_rank_iterable)
              re_second_epoch = []
              for _ in range(forward_steps):
                  re_second_epoch.append(next(iterator))
              assert re_second_epoch == all_rank_second_epoch[rank]
  
imports: import numpy
  import pytest
  from functools import partial
  from itertools import chain
  from copy import deepcopy
  from fastNLP.core.samplers.reproducible_sampler import RandomSampler, SortedSampler, SequentialSampler
  from tests.helpers.datasets.torch_data import TorchNormalDataset
label_desc: Use image node_id 4 for calling the TestRandomSampler obj's underlying member method code with example usage: obj.test_4() without return types