class_docstr: Manage shuffle execution for ray by resolve dependencies between mappers outputs and reducers inputs based on
  mapper and reducer index.
code: def __init__(self, subtask_graph: SubtaskGraph):
      self._subtask_graph = subtask_graph
      self._proxy_subtasks = subtask_graph.get_shuffle_proxy_subtasks()
      self._num_shuffles = subtask_graph.num_shuffles()
      self._mapper_output_refs = []
      self._mapper_indices = {}
      self._reducer_indices = {}
      for shuffle_index, proxy_subtask in enumerate(
          self._proxy_subtasks
      ):
          # Note that the reducers can also be mappers such as `DuplicateOperand`.
          mapper_subtasks = subtask_graph.predecessors(proxy_subtask)
          reducer_subtasks = subtask_graph.successors(proxy_subtask)
          n_mappers = len(mapper_subtasks)
          n_reducers = proxy_subtask.chunk_graph.results[
              0
          ].op.n_reducers
          mapper_output_arr = np.empty(
              (n_mappers, n_reducers), dtype=object
          )
          self._mapper_output_refs.append(mapper_output_arr)
          self._mapper_indices.update(
              {
                  subtask: (shuffle_index, mapper_index)
                  for mapper_index, subtask in enumerate(
                      mapper_subtasks
                  )
              }
          )
          # reducers subtask should be sorted by reducer_index and MapReduceOperand.map should insert shuffle block
          # in reducers order, otherwise shuffle blocks will be sent to wrong reducers.
          sorted_filled_reducer_subtasks = (
              self._get_sorted_filled_reducers(
                  reducer_subtasks, n_reducers
              )
          )
          self._reducer_indices.update(
              {
                  subtask: (shuffle_index, reducer_ordinal)
                  for reducer_ordinal, subtask in enumerate(
                      sorted_filled_reducer_subtasks
                  )
              }
          )
  
file_path: mars/mars/services/task/execution/ray/shuffle.py
imports: import numpy
  from typing import List, Iterable
  from .....core.operand import MapReduceOperand, OperandStage
  from .....utils import lazy_import
  from ....subtask import Subtask, SubtaskGraph
label_desc: Use image node_id 1 to create a new ShuffleManager object  with example: obj = ShuffleManager(subtask_graph)