code: def test_local_shuffle(ray_start_regular_shared):
      # confirm that no data disappears, and they all stay within the same shard
      it = from_range(8, num_shards=2).local_shuffle(
          shuffle_buffer_size=2
      )
      assert repr(it) == (
          "ParallelIterator[from_range[8, shards=2]"
          + ".local_shuffle(shuffle_buffer_size=2, seed=None)]"
      )
      shard_0 = it.get_shard(0)
      shard_1 = it.get_shard(1)
      assert set(shard_0) == {0, 1, 2, 3}
      assert set(shard_1) == {4, 5, 6, 7}
  
      # check that shuffling results in different orders
      it1 = from_range(100, num_shards=10).local_shuffle(
          shuffle_buffer_size=5
      )
      it2 = from_range(100, num_shards=10).local_shuffle(
          shuffle_buffer_size=5
      )
      assert list(it1.gather_sync()) != list(it2.gather_sync())
  
      # buffer size of 1 should not result in any shuffling
      it3 = from_range(10, num_shards=1).local_shuffle(
          shuffle_buffer_size=1
      )
      assert list(it3.gather_sync()) == list(range(10))
  
      # statistical test
      it4 = from_items([0, 1] * 10000, num_shards=1).local_shuffle(
          shuffle_buffer_size=100
      )
      result = "".join(it4.gather_sync().for_each(str))
      freq_counter = Counter(zip(result[:-1], result[1:]))
      assert len(freq_counter) == 4
      for key, value in freq_counter.items():
          assert value / len(freq_counter) > 0.2
  
imports: import sys
  import time
  import collections
  from collections import Counter
  import pytest
  import ray
  from ray.util.iter import from_items, from_iterators, from_range, from_actors, ParallelIteratorWorker, LocalIterator
  from ray._private.test_utils import Semaphore
label_desc: Use image node_id 19 for calling a global function with example usage: test_local_shuffle(ray_start_regular_shared) without return types