docstr: When there are multiple nodes, merge tasks are spread throughout the
  cluster to improve load-balancing. Each merge task produces outputs for
  a contiguous partition of reduce tasks. This method creates an iterator
  that returns reduce task indices round-robin across the merge tasks.
  This can be used to submit reduce tasks in a way that spreads the load
  evenly across the cluster.
docstr_tok: ["When","there","are","multiple","nodes",",","merge","tasks","are","spread","throughout","the","cluster","to","improve","load-balancing",".","Each","merge","task","produces","outputs","for","a","contiguous","partition","of","reduce","tasks",".","This","method","creates","an","iterator","that","returns","reduce","task","indices","round-robin","across","the","merge","tasks",".","This","can","be","used","to","submit","reduce","tasks","in","a","way","that","spreads","the","load","evenly","across","the","cluster","."]
code: def round_robin_reduce_idx_iterator(self):
      """
      When there are multiple nodes, merge tasks are spread throughout the
      cluster to improve load-balancing. Each merge task produces outputs for
      a contiguous partition of reduce tasks. This method creates an iterator
      that returns reduce task indices round-robin across the merge tasks.
      This can be used to submit reduce tasks in a way that spreads the load
      evenly across the cluster.
      """
      idx = 0
      round_idx = 0
      while idx < self.output_num_blocks:
          for merge_idx in range(self.num_merge_tasks_per_round):
              if merge_idx < self._partitions_with_extra_task:
                  reduce_idx = merge_idx * (
                      self.merge_partition_size + 1
                  )
                  partition_size = self.merge_partition_size + 1
              else:
                  reduce_idx = self._partitions_with_extra_task * (
                      self.merge_partition_size + 1
                  )
                  merge_idx -= self._partitions_with_extra_task
                  reduce_idx += merge_idx * self.merge_partition_size
                  partition_size = self.merge_partition_size
  
              if round_idx >= partition_size:
                  continue
  
              reduce_idx += round_idx
              yield reduce_idx
              idx += 1
          round_idx += 1
  
imports: import logging
  import math
  from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.shuffle import ShuffleOp
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 4 for calling the _MergeTaskSchedule obj's underlying member method code with example usage: obj.round_robin_reduce_idx_iterator() without return types