class_docstr: An subtask graph builder which build subtask graph for chunk graph based on passed band_resource.
  
  If push shuffle is used, this builder will validate predecessors orders consistency of ShuffleProxy between
  chunk graph and generated subtask graph.
code: def _gen_subtask_info(
      self,
      chunks: List[ChunkType],
      chunk_to_subtask: Dict[ChunkType, Subtask],
      chunk_to_bands: Dict[ChunkType, BandType],
      chunk_to_fetch_chunk: Dict[ChunkType, ChunkType],
  ) -> Tuple[Subtask, List[Subtask], bool]:
      # gen subtask and its input subtasks
      chunks_set = set(chunks)
      result_chunks = []
      result_chunks_set = set()
      chunk_graph = ChunkGraph(result_chunks)
      out_of_scope_chunks = []
      chunk_to_copied = self._chunk_to_copied
      update_meta_chunks = []
      # subtask properties
      band = None
      is_virtual = None
      retryable = True
      chunk_priority = None
      expect_band = None
      bands_specified = None
      processed = set()
      for chunk in chunks:
          if chunk in processed:
              continue
          if expect_band is None:
              expect_band = self._get_expect_band(chunk.op)
              bands_specified = expect_band is not None
          else:  # pragma: no cover
              curr_expect_band = self._get_expect_band(chunk.op)
              assert (
                  curr_expect_band is None
                  or expect_band == curr_expect_band
              ), (
                  f"expect_band {curr_expect_band} conflicts with chunks that have same color: "
                  f"{expect_band}"
              )
          # process band
          chunk_band = chunk_to_bands.get(chunk)
          if chunk_band is not None:
              assert (
                  band is None or band == chunk_band
              ), "band conflicts with chunks that have same color"
              band = chunk_band
          # process is_virtual
          if isinstance(chunk.op, VirtualOperand):
              assert (
                  is_virtual is None
              ), "only 1 virtual operand can exist"
              is_virtual = True
          else:
              is_virtual = False
          # process retryable
          if not chunk.op.retryable:
              retryable = False
          # process priority
          if chunk.op.priority is not None:
              assert (
                  chunk_priority is None
                  or chunk_priority == chunk.op.priority
              ), "priority conflicts with chunks that have same color"
              chunk_priority = chunk.op.priority
          # process input chunks
          inp_chunks = []
          build_fetch_index_to_chunks = dict()
          for i, inp_chunk in enumerate(chunk.inputs):
              if inp_chunk in chunks_set:
                  inp_chunks.append(chunk_to_copied[inp_chunk])
              else:
                  build_fetch_index_to_chunks[i] = inp_chunk
                  inp_chunks.append(None)
                  if not isinstance(inp_chunk.op, Fetch):
                      out_of_scope_chunks.append(inp_chunk)
          fetch_chunks = self._gen_input_chunks(
              list(build_fetch_index_to_chunks.values()),
              chunk_to_fetch_chunk,
          )
          for i, fetch_chunk in zip(
              build_fetch_index_to_chunks, fetch_chunks
          ):
              inp_chunks[i] = fetch_chunk
          copied_op = chunk.op.copy()
          copied_op._key = chunk.op.key
          out_chunks = [
              c.data
              for c in copied_op.new_chunks(
                  inp_chunks,
                  kws=[c.params.copy() for c in chunk.op.outputs],
              )
          ]
          for src_chunk, out_chunk in zip(chunk.op.outputs, out_chunks):
              processed.add(src_chunk)
              out_chunk._key = src_chunk.key
              chunk_graph.add_node(out_chunk)
              # cannot be copied twice
              assert src_chunk not in chunk_to_copied
              chunk_to_copied[src_chunk] = out_chunk
              if src_chunk in self._final_result_chunks_set:
                  if out_chunk not in result_chunks_set:
                      # add to result chunks
                      result_chunks.append(out_chunk)
                      # chunk is in the result chunks of full chunk graph
                      # meta need to be updated
                      update_meta_chunks.append(out_chunk)
                      result_chunks_set.add(out_chunk)
              if not is_virtual:
                  # skip adding fetch chunk to chunk graph when op is virtual operand
                  for c in inp_chunks:
                      if c not in chunk_graph:
                          chunk_graph.add_node(c)
                      chunk_graph.add_edge(c, out_chunk)
      stage_n_outputs = len(result_chunks)
      # add chunks with no successors into result chunks
      result_chunks.extend(
          c
          for c in chunk_graph.iter_indep(reverse=True)
          if c not in result_chunks_set
      )
      expect_bands = (
          [expect_band]
          if bands_specified
          else ([band] if band is not None else None)
      )
      # calculate priority
      if out_of_scope_chunks:
          inp_subtasks = []
          for out_of_scope_chunk in out_of_scope_chunks:
              copied_out_of_scope_chunk = chunk_to_copied[
                  out_of_scope_chunk
              ]
              inp_subtask = chunk_to_subtask[out_of_scope_chunk]
              if (
                  copied_out_of_scope_chunk
                  not in inp_subtask.chunk_graph.result_chunks
              ):
                  # make sure the chunk that out of scope
                  # is in the input subtask's results,
                  # or the meta may be lost
                  inp_subtask.chunk_graph.result_chunks.append(
                      copied_out_of_scope_chunk
                  )
              inp_subtasks.append(inp_subtask)
          depth = max(st.priority[0] for st in inp_subtasks) + 1
      else:
          inp_subtasks = []
          depth = 0
      priority = (depth, chunk_priority or 0)
  
      subtask = Subtask(
          subtask_id=new_task_id(),
          stage_id=self._stage_id,
          logic_key=self._gen_logic_key(chunks),
          session_id=self._task.session_id,
          task_id=self._task.task_id,
          chunk_graph=chunk_graph,
          expect_bands=expect_bands,
          bands_specified=bands_specified,
          virtual=is_virtual,
          priority=priority,
          retryable=retryable,
          update_meta_chunks=update_meta_chunks,
          extra_config=self._extra_config,
          stage_n_outputs=stage_n_outputs,
      )
  
      is_shuffle_proxy = False
      if self._has_shuffle:
          proxy_chunks = [
              c for c in result_chunks if isinstance(c.op, ShuffleProxy)
          ]
          if proxy_chunks:
              assert len(proxy_chunks) <= 1, proxy_chunks
              is_shuffle_proxy = True
      return subtask, inp_subtasks, is_shuffle_proxy
  
imports: import itertools
  import logging
  from collections import deque, defaultdict
  from typing import Dict, List, Tuple, Type, Union
  from ....config import Config
  from ....core import ChunkGraph, ChunkType, enter_mode, DAG
  from ....core.operand import Fetch, VirtualOperand, LogicKeyGenerator, MapReduceOperand, OperandStage, ShuffleProxy, ShuffleFetchType
  from ....lib.ordered_set import OrderedSet
  from ....resource import Resource
  from ....typing import BandType, OperandType
  from ....utils import build_fetch, build_fetch_shuffle, tokenize
  from ...subtask import SubtaskGraph, Subtask
  from ..core import Task, new_task_id, MapReduceInfo
  from .assigner import AbstractGraphAssigner, GraphAssigner
  from .fusion import Coloring
label_desc: Use image node_id 7 for calling the GraphAnalyzer obj's underlying member method code with example usage: obj._gen_subtask_info(chunks, chunk_to_subtask, chunk_to_bands, chunk_to_fetch_chunk) and returns: subtask,  inp_subtasks,  is_shuffle_proxy