code: def _construct_script(
      task_spec: common.TaskSpec,
      globals: GlobalsBridge,
      *,
      number: int,
      repeats: int,
      collect_baseline: bool,
      error_log: str,
      stat_log: str,
      bindings: Optional[CallgrindModuleType],
  ) -> str:
      def block_stmt(stmt: str, indent: int = 0) -> str:
          """Partially unroll benchmark loop.
  
          The naive template looks something like:
              "for _ in range({number}): {stmt}"
  
          However a loop in Python is surprisingly expensive, and significantly
          increases the number of background Python instructions. So instead we
          partially unroll the loops, with a block size of 100 chosen to keep
          the instruction overhead from `range` low while also not ballooning
          the size of the generated file.
          """
          block_size = 100
          loop_count = number // block_size
          if loop_count == 1:
              # There is no point in having `for _ in range(1): ...` rather
              # than just `...`, and this lets us save shave a few background
              # instructions.
              loop_count = 0
          remainder = number - block_size * loop_count
          blocked_stmt = ""
  
          if loop_count:
              unrolled_stmts = textwrap.indent(
                  "\n".join([stmt] * block_size), " " * 4
              )
              blocked_stmt += (
                  f"for _ in range({loop_count}):\n{unrolled_stmts}\n"
              )
  
          if remainder:
              blocked_stmt += "\n".join([stmt] * remainder)
  
          return textwrap.indent(blocked_stmt, " " * indent)
  
      pass_baseline = (
          "callgrind_bindings._valgrind_toggle()\n"
          f"{block_stmt('pass')}\n"
          "callgrind_bindings._valgrind_toggle_and_dump_stats()"
      )
  
      return (
          textwrap.dedent(
              r"""
              import gc
              import os
              import pickle
              import subprocess
              import sys
              import time
  
              # Mitigate https://github.com/pytorch/pytorch/issues/37377
              # which can sometimes cause the subprocess call to fail.
              import numpy as np
  
              import torch
              torch.set_num_threads({num_threads})
  
              {bindings_import}
  
              PID = os.getpid()
  
              def log_failure(msg):
                  with open({error_log_repr}, "wt") as f:
                      f.write(msg)
                  sys.exit(1)
  
              def check_result(completed_process):
                  if completed_process.returncode:
                      log_failure(f"Command failed: {{' '.join(completed_process.args)}}")
                  return completed_process
  
              # =============================================================================
              # == Check that subprocess matches parent =====================================
              # =============================================================================
              if os.path.realpath(sys.executable) != "{parent_interpreter}":
                  log_failure(
                      "Interpreter mismatch:\n"
                      f"  {{os.path.realpath(sys.executable)}}\n    vs.\n  {parent_interpreter}"
                  )
  
              if torch.__file__ != "{torch_file}":
                  log_failure(
                      "PyTorch does not match expected file:\n"
                      f"  {{torch.__file__}}\n    vs.\n  {torch_file}"
                  )
  
              # =============================================================================
              # == User specified setup =====================================================
              # =============================================================================
              # Load serialized globals
              {load_globals}
  
              # User setup str
              {setup}
  
              for _ in range({warmup_number}):
              {indented_stmt}
  
              # =============================================================================
              # == Callgrind management =====================================================
              # =============================================================================
              with open("{stat_log}", "wb") as stat_file:
                  # If many instances of callgrind are running at once, the output of
                  # `callgrind_control` may exceed 16kb which would cause `subprocess.PIPE`
                  # to deadlock. So instead we use a file.
                  callgrind_stat = check_result(subprocess.run(
                      ["callgrind_control", "--stat"],
                      stdout=stat_file,
                      stderr=subprocess.STDOUT,
                  ))
  
              with open("{stat_log}", "rt") as stat_file:
                  stat_lines = stat_file.read().splitlines()
  
              if f"PID {{PID}}: python {{__file__}}" not in stat_lines:
                  log_failure("Process does not appear to be running callgrind.")
  
              gc.collect()
              time.sleep(0.01)
  
              # =============================================================================
              # == User code block ==========================================================
              # =============================================================================
              for _ in range({repeats}):
                  callgrind_bindings._valgrind_toggle()
              {blocked_stmt}
                  callgrind_bindings._valgrind_toggle_and_dump_stats()
                  gc.collect()
  
              {baseline}
          """
          )
          .strip()
          .format(
              indented_stmt=textwrap.indent(task_spec.stmt, " " * 4),
              blocked_stmt=block_stmt(task_spec.stmt, indent=4),
              baseline=(pass_baseline if collect_baseline else ""),
              number=number,
              repeats=repeats,
              load_globals=globals.construct(),
              setup=task_spec.setup,
              warmup_number=min(number, 10),
              num_threads=task_spec.num_threads,
              error_log_repr=repr(error_log),
              stat_log=stat_log,
              parent_interpreter=os.path.realpath(sys.executable),
              torch_file=torch.__file__,
              bindings_import=(
                  "import torch._C as callgrind_bindings"
                  if bindings is None
                  else f"import {bindings.__name__} as callgrind_bindings"
              ),
          )
      )
  
imports: import collections
  import enum
  import dataclasses
  import itertools
  import os
  import pickle
  import re
  import shutil
  import subprocess
  import sys
  import textwrap
  from typing import cast, Any, Callable, DefaultDict, Dict, Iterator, List, NamedTuple, Optional, Tuple, Union, TYPE_CHECKING
  import torch
  from torch.utils.benchmark.utils import common, cpp_jit
  from torch.utils.benchmark.utils._stubs import CallgrindModuleType
  import operator
label_desc: Use image node_id 5 for calling the _ValgrindWrapper obj's underlying member method code with example usage: obj._construct_script(task_spec, globals) and returns: textwrap, textwrap