code: def _get_subtask_out_info(
      subtask_chunk_graph: ChunkGraph,
      is_mapper: bool,
      n_reducers: int = None,
  ):
      # output_keys might be duplicate in chunk graph, use dict to deduplicate.
      # output_keys order should be consistent with remote `execute_subtask`,
      # dict can preserve insert order.
      output_keys = {}
      shuffle_chunk = None
      if is_mapper:
          assert n_reducers is not None
          if len(subtask_chunk_graph.result_chunks) == 1:
              return set(), n_reducers
          for chunk in subtask_chunk_graph.result_chunks:
              if not chunk.is_mapper:
                  output_keys[chunk.key] = 1
                  # mapper may produce outputs which isn't shuffle blocks, such as TensorUnique._execute_agg_reduce
                  # which is  mapper too, but some outputs are not mapper blocks:
                  # https://user-images.githubusercontent.com/12445254/184132642-a19259fd-43d6-4a27-a033-4aaa97d7586e.svg
              else:
                  assert shuffle_chunk is None, (shuffle_chunk, chunk)
                  shuffle_chunk = chunk
          return output_keys.keys(), len(output_keys) + n_reducers
      for chunk in subtask_chunk_graph.result_chunks:
          if isinstance(
              chunk.op, VirtualOperand
          ):  # FIXME(chaokunyang) no need to check this?
              continue
          else:
              output_keys[chunk.key] = 1
      return output_keys.keys(), len(output_keys)
  
imports: import asyncio
  import collections
  import enum
  import functools
  import itertools
  import logging
  import operator
  import time
  from dataclasses import dataclass, field
  from typing import List, Dict, Any, Callable
  import numpy
  from .....core import ChunkGraph, Chunk, TileContext
  from .....core.context import set_context
  from .....core.operand import Fetch, Fuse, VirtualOperand, execute
  from .....core.operand.fetch import FetchShuffle
  from .....lib.aio import alru_cache
  from .....metrics.api import init_metrics, Metrics
  from .....resource import Resource
  from .....serialization import serialize, deserialize
  from .....typing import BandType
  from .....utils import aiotask_wrapper, calc_data_size, lazy_import, get_chunk_params
  from ....lifecycle.api import LifecycleAPI
  from ....meta.api import MetaAPI
  from ....subtask import Subtask, SubtaskGraph
  from ....subtask.utils import iter_output_data
  from ...core import Task
  from ..api import TaskExecutor, ExecutionChunkResult, register_executor_cls
  from ..utils import ResultTileablesLifecycle
  from .config import RayExecutionConfig, IN_RAY_CI
  from .context import RayExecutionContext, RayExecutionWorkerContext, RayRemoteObjectManager
  from .shuffle import ShuffleManager
label_desc: Use image node_id 4 for calling a global function with example usage: _get_subtask_out_info(subtask_chunk_graph, is_mapper, n_reducers) and returns: output_keys,  len, output_keys,  unknown, set,  n_reducers