code: def _export_subtask_to_dot(
      cls,
      subtask: Subtask,
      subgraph_name: str,
      current_stage: int,
      line_colors: Dict,
      chunk_key_to_subtask: Dict[str, List],
      trunc_key: int = 5,
  ):
      chunk_graph = subtask.chunk_graph
      sio = StringIO()
      chunk_style = "[shape=box]"
      operand_style = "[shape=circle]"
  
      visited = set()
      all_nodes = []
      for node in chunk_graph.iter_nodes():
          op = node.op
          if isinstance(node.op, (Fetch, FetchShuffle)):
              continue
          op_name = type(op).__name__
          if op.stage is not None:
              op_name = f"{op_name}:{op.stage.name}"
          if op.key in visited:
              continue
          for input_chunk in op.inputs or []:
              if input_chunk.key not in visited and not isinstance(
                  input_chunk.op, (Fetch, FetchShuffle)
              ):  # pragma: no cover
                  node_name = f'"Chunk:{cls._gen_chunk_key(input_chunk, trunc_key)}"'
                  sio.write(f"{node_name} {chunk_style}\n")
                  all_nodes.append(node_name)
                  visited.add(input_chunk.key)
              if op.key not in visited:
                  node_name = f'"{op_name}:{op.key[:trunc_key]}"'
                  sio.write(f"{node_name} {operand_style}\n")
                  all_nodes.append(node_name)
                  visited.add(op.key)
              if (
                  isinstance(input_chunk.op, (Fetch, FetchShuffle))
                  and input_chunk.key in chunk_key_to_subtask
              ):
                  stage, tail_cluster = chunk_key_to_subtask[
                      input_chunk.key
                  ]
                  if stage == current_stage:
                      line_style = "style=bold"
                  else:
                      line_style = f"style=bold color={line_colors[(current_stage, stage)]}"
                  sio.write(
                      f'"Chunk:{cls._gen_chunk_key(input_chunk, trunc_key)}" ->'
                      f' "{op_name}:{op.key[:trunc_key]}" '
                      f"[lhead={subgraph_name} ltail={tail_cluster} {line_style}];\n"
                  )
              else:
                  sio.write(
                      f'"Chunk:{cls._gen_chunk_key(input_chunk, trunc_key)}" -> '
                      f'"{op_name}:{op.key[:trunc_key]}"\n'
                  )
  
          for output_chunk in op.outputs or []:
              if output_chunk.key not in visited:
                  node_name = f'"Chunk:{cls._gen_chunk_key(output_chunk, trunc_key)}"'
                  sio.write(f"{node_name} {chunk_style}\n")
                  all_nodes.append(node_name)
                  visited.add(output_chunk.key)
              if op.key not in visited:
                  node_name = f'"{op_name}:{op.key[:trunc_key]}"'
                  sio.write(f"{node_name} {operand_style}\n")
                  all_nodes.append(node_name)
                  visited.add(op.key)
              sio.write(
                  f'"{op_name}:{op.key[:trunc_key]}" -> '
                  f'"Chunk:{cls._gen_chunk_key(output_chunk, trunc_key)}"\n'
              )
      # write subgraph info
      sio.write(f"subgraph {subgraph_name} {{\n")
      nodes_str = " ".join(all_nodes)
      sio.write(f"{nodes_str};\n")
      sio.write(f'label="{subtask.subtask_id}";\n}}')
      sio.write("\n")
      return sio.getvalue()
  
file_path: mars/mars/services/task/supervisor/graph_visualizer.py
imports: import itertools
  from io import StringIO
  from typing import Dict, List
  from ....core.operand import Fetch, FetchShuffle
  from ...subtask import Subtask, SubtaskGraph
label_desc: Use image node_id 3 for calling the GraphVisualizer obj's underlying member method code with example usage: obj._export_subtask_to_dot(cls, subtask, subgraph_name, current_stage, line_colors, chunk_key_to_subtask, trunc_key) and returns: sio