docstr: Check shuffle reduce chunks sorted reducer_index consistent with reducer_ordinal. So shuffle mapper blocks
  can be sorted by reducer_index, and the reducer can fetch mapper data by reducer_ordinal.
docstr_tok: ["Check","shuffle","reduce","chunks","sorted","reducer_index","consistent","with","reducer_ordinal",".","So","shuffle","mapper","blocks","can","be","sorted","by","reducer_index",",","and","the","reducer","can","fetch","mapper","data","by","reducer_ordinal","."]
code: def _check_shuffle_reduce_chunks(cls, chunks: List, checked_chunks):
      """Check shuffle reduce chunks sorted reducer_index consistent with reducer_ordinal. So shuffle mapper blocks
      can be sorted by reducer_index, and the reducer can fetch mapper data by reducer_ordinal.
      """
      chunks = [c for c in chunks or [] if c not in checked_chunks]
      if not chunks:
          return
      from .....core.operand import (
          MapReduceOperand,
          ShuffleProxy,
          OperandStage,
      )
  
      reduce_chunks = defaultdict(list)
      for c in chunks:
          checked_chunks.add(c)
          if (
              isinstance(c.op, MapReduceOperand)
              and c.op.stage == OperandStage.reduce
          ):
              shuffle_proxies = [
                  c for c in c.inputs if isinstance(c.op, ShuffleProxy)
              ]
              assert len(shuffle_proxies) == 1, (
                  c.inputs,
                  shuffle_proxies,
              )
              reduce_chunks[shuffle_proxies[0]].append(c)
          else:
              cls._check_shuffle_reduce_chunks(c.inputs, checked_chunks)
      for _, reduce_chunks in reduce_chunks.items():
          sorted_chunks_by_indices = sorted(
              reduce_chunks, key=lambda c: c.op.reducer_index
          )
          sorted_chunks_by_ordinals = sorted(
              reduce_chunks, key=lambda c: c.op.reducer_ordinal
          )
          for c1, c2 in zip(
              sorted_chunks_by_indices, sorted_chunks_by_ordinals
          ):
              assert c1.op.reducer_index == c2.op.reducer_index, (
                  sorted_chunks_by_indices,
                  sorted_chunks_by_ordinals,
              )
          for c in reduce_chunks:
              cls._check_shuffle_reduce_chunks(c.inputs, checked_chunks)
  
imports: import itertools
  from collections import defaultdict
  from functools import partial
  from typing import Callable, Dict, List
  import numpy
  from .....core import TileableType, ChunkGraph, OBJECT_TYPE, enter_mode, register, unregister
  from .....core.operand import Fetch, ShuffleProxy
  from .....core.operand.shuffle import ShuffleFetchType
  from .....resource import Resource
  from .....tests.core import _check_args, ObjectCheckMixin
  from .....typing import BandType, ChunkType
  from ....subtask import Subtask, SubtaskGraph
  from ...analyzer import GraphAnalyzer
  from ..preprocessor import CancellableTiler, TaskPreprocessor
label_desc: Use image node_id 8 for calling the CheckedTaskPreprocessor obj's underlying member method code with example usage: obj._check_shuffle_reduce_chunks(cls, chunks, checked_chunks) without return types