code: def execute_subtask(
      subtask_id: str,
      subtask_chunk_graph: ChunkGraph,
      output_meta_n_keys: int,
      is_mapper,
      *inputs,
  ):
      """
      The function used for execute subtask in ray task.
  
      Parameters
      ----------
      subtask_id: str
          id of subtask
      subtask_chunk_graph: ChunkGraph
          chunk graph for subtask
      output_meta_n_keys: int
          will be 0 if subtask is a shuffle mapper.
      is_mapper: bool
          Whether current subtask is a shuffle mapper. Note that shuffle reducers such as `DataFrameDropDuplicates`
          can be a mapper at the same time.
      inputs:
          inputs for current subtask
  
      Returns
      -------
          subtask outputs and meta for outputs if `output_meta_keys` is provided.
      """
      init_metrics("ray")
      started_subtask_number.record(1)
      ray_task_id = ray.get_runtime_context().get_task_id()
      subtask_chunk_graph = deserialize(*subtask_chunk_graph)
      logger.info(
          "Start subtask: %s, ray task id: %s.", subtask_id, ray_task_id
      )
      # Optimize chunk graph.
      subtask_chunk_graph = _optimize_subtask_graph(subtask_chunk_graph)
      fetch_chunks, shuffle_fetch_chunk = _get_fetch_chunks(
          subtask_chunk_graph
      )
      context = RayExecutionWorkerContext(RayTaskState.get_handle)
      if shuffle_fetch_chunk is not None:
          # The subtask is a reducer subtask.
          n_mappers = shuffle_fetch_chunk.op.n_mappers
          # Some reducer may have multiple output chunks, see `PSRSshuffle._execute_reduce` and
          # https://user-images.githubusercontent.com/12445254/168569524-f09e42a7-653a-4102-bdf0-cc1631b3168d.png
          reducer_chunks = subtask_chunk_graph.successors(
              shuffle_fetch_chunk
          )
          reducer_operands = set(c.op for c in reducer_chunks)
          if len(reducer_operands) != 1:  # pragma: no cover
              raise ValueError(
                  f"Subtask {subtask_id} has more than 1 reduce operands: {subtask_chunk_graph.to_dot()}"
              )
          reducer_operand = reducer_chunks[0].op
          reducer_index = reducer_operand.reducer_index
          # Virtual shuffle keys, keep this in sync with `MapReducerOperand#_iter_mapper_key_idx_pairs`
          context.update(
              {
                  (i, reducer_index): block
                  for i, block in enumerate(inputs[-n_mappers:])
              }
          )
          inputs = inputs[:-n_mappers]
      shuffle_input_key_count = len(context)
      # Create a subtask GC object.
      subtask_gc = _SubtaskGC(subtask_chunk_graph, context)
      # Update non shuffle inputs to context.
      context.update(
          zip((start_chunk.key for start_chunk in fetch_chunks), inputs)
      )
  
      for chunk in subtask_chunk_graph.topological_iter():
          if chunk.key not in context:
              try:
                  context.set_current_chunk(chunk)
                  execute(context, chunk.op)
              except Exception:
                  logger.exception(
                      "Execute operand %s of graph %s failed.",
                      chunk.op,
                      subtask_chunk_graph.to_dot(),
                  )
                  raise
          subtask_gc.gc_inputs(chunk)
  
      # For non-mapper subtask, output context is chunk key to results.
      # For mapper subtasks, output context is data key to results.
      # `iter_output_data` must ensure values order since we only return values.
      normal_output = {}
      mapper_output = {}
      for key, data, is_mapper_block in iter_output_data(
          subtask_chunk_graph, context
      ):
          if is_mapper_block:
              mapper_output[key] = data
          else:
              normal_output[key] = data
  
      # The inputs are referenced by the Ray worker in _raylet.pyx, GC them in Mars is useless.
      # So, subtask GC has skipped GC shuffle input keys in order to simplify the implementation.
      expect_context_count = (
          len(normal_output)
          + len(mapper_output)
          + shuffle_input_key_count
      )
      assert (
          len(context) == expect_context_count
      ), f"The remaining context count mismatch: {len(context)}(actual) != {expect_context_count}(expected)."
  
      output_values = []
      # assert output keys order consistent
      if is_mapper:
          # mapper may produce outputs which isn't shuffle blocks, such as TensorUnique._execute_agg_reduce.
          mapper_main_keys = set(k[0] for k in mapper_output.keys())
          assert len(mapper_main_keys) == 1, mapper_main_keys
          # sorted reducer_index's consistency with reducer_ordinal is checked in
          # `OperandTilesHandler._check_shuffle_reduce_chunks`.
          # So sort keys by reducer_index to ensure mapper outputs consist with reducer_ordinal,
          # then downstream can fetch shuffle blocks by reducer_ordinal.
          mapper_output = dict(
              sorted(mapper_output.items(), key=lambda item: item[0][1])
          )
      if output_meta_n_keys:
          output_meta = {}
          # for non-shuffle subtask, record meta in supervisor.
          for chunk in subtask_chunk_graph.result_chunks[
              :output_meta_n_keys
          ]:
              chunk_key = chunk.key
              if chunk_key not in output_meta:
                  if isinstance(chunk.op, Fuse):  # pragma: no cover
                      # fuse op
                      chunk = chunk.chunk
                  data = context[chunk_key]
                  memory_size = calc_data_size(data)
                  output_meta[chunk_key] = (
                      get_chunk_params(chunk),
                      memory_size,
                  )
          output_values.append(output_meta)
      output_values.extend(normal_output.values())
      output_values.extend(mapper_output.values())
      logger.info(
          "Complete subtask: %s, ray task id: %s.",
          subtask_id,
          ray_task_id,
      )
      completed_subtask_number.record(1)
      return (
          output_values[0] if len(output_values) == 1 else output_values
      )
  
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 2 for calling a global function with example usage: execute_subtask(subtask_id, subtask_chunk_graph, output_meta_n_keys, is_mapper) and returns: unknown