code: def _run_benchmarks(
      metadata: BenchmarkMetadata,
      shard: ShardingStrategy,
      suite: BenchmarkSuite,
      dest: Path,
      cache_dir: Path,
  ) -> Optional[pd.DataFrame]:
      data_cache_dir = cache_dir / "data"
  
      # Run all tasks in separate processes, to make sure all memory etc. is released.
      mp_ctx = mp.get_context("spawn")
  
      tasks = suite.get_tasks()
      global_repetition = 0
      total_repetitions = sum(t.repetitions for t in tasks)
      max_task_repetitions = max(t.repetitions for t in tasks)
  
      metrics_list = []
      metrics_df: Optional[pd.DataFrame] = None
  
      t_before = perf_counter()
      for task_repetition in range(max_task_repetitions):
          for task in tasks:
              if task_repetition >= task.repetitions:
                  continue
  
              if global_repetition % shard.shard_n != shard.shard_i:
                  global_repetition += 1
                  continue
  
              for retry in range(3):
                  t_now = perf_counter()
                  fraction_done = global_repetition / total_repetitions
                  if global_repetition == 0:
                      eta_str = "???"
                  else:
                      velocity = (t_now - t_before) / global_repetition
                      t_end = velocity * total_repetitions + t_before
                      t_remaining = t_end - t_now
                      eta_str = f"{timedelta(seconds=t_remaining)}"
                  print(
                      "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
                  )
                  print(
                      f"Running: {task.name}/rep={task_repetition} (retry {retry})"
                  )
                  print(f"{fraction_done: .2%} done. ETA: {eta_str}.")
                  print(
                      "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
                  )
  
                  random_seed = (
                      BASE_SEED
                      + task_repetition
                      + retry * task.repetitions
                  )
                  metrics_queue: "mp.Queue[Optional[Mapping[Metric, float]]]" = (
                      mp_ctx.Queue()
                  )
                  metrics_process = mp_ctx.Process(
                      target=_collect_metrics_process,
                      args=(
                          task,
                          data_cache_dir,
                          random_seed,
                          metrics_queue,
                      ),
                  )
                  metrics_process.start()
                  metrics = metrics_queue.get(timeout=ONE_HOUR_S)
                  metrics_process.join()
                  assert metrics_process.exitcode == 0
                  if metrics is None:
                      print(
                          "Model failed. Retrying with another random seed."
                      )
                      continue
  
                  for metric, metric_value in metrics.items():
                      metrics_list.append(
                          (
                              metadata.run_id,
                              task.dataset_name,
                              task.model_name,
                              task.do_compile,
                              task.do_optimise,
                              task_repetition,
                              metric.name,
                              metric_value,
                          )
                      )
                  metrics_df = pd.DataFrame(
                      metrics_list,
                      columns=[
                          "run_id",
                          "dataset",
                          "model",
                          "do_compile",
                          "do_optimise",
                          "repetition",
                          "metric",
                          "value",
                      ],
                  )
                  print(
                      tabulate(
                          metrics_df, headers="keys", showindex="never"
                      )
                  )
                  metrics_path = (
                      dest / f"metrics{shard.file_suffix}.csv"
                  )
                  metrics_df.to_csv(metrics_path, index=False)
                  break
              global_repetition += 1
  
      return metrics_df
  
imports: import argparse
  import json
  import multiprocessing
  import traceback
  from datetime import timedelta
  from pathlib import Path
  from time import perf_counter
  from typing import Mapping, Optional
  import numpy
  import pandas
  from tabulate import tabulate
  import benchmark.benchmarks
  import benchmark.metrics
  import gpflow
  from benchmark.benchmark_api import BENCHMARK_SUITES, BenchmarkSuite, BenchmarkTask
  from benchmark.dataset_api import DATASET_FACTORIES
  from benchmark.metadata import BenchmarkMetadata
  from benchmark.metric_api import Metric
  from benchmark.model_api import MODEL_FACTORIES
  from benchmark.paths import setup_dest
  from benchmark.plot import plot
  from benchmark.sharding import ShardingStrategy
label_desc: Use image node_id 3 for calling a global function with example usage: _run_benchmarks(metadata, shard, suite, dest, cache_dir) and returns: metrics_df