code: def test_warmup_checkpointing(self) -> None:
      warmup_steps = 5
      extra_steps = 2
      batch_size = 128
      qps = TowerQPSMetric(
          world_size=1,
          my_rank=0,
          batch_size=batch_size,
          tasks=[DefaultTaskInfo],
          warmup_steps=warmup_steps,
          compute_on_all_ranks=False,
          should_validate_update=False,
          window_size=200,
      )
      model_output = gen_test_batch(batch_size)
      for i in range(5):
          for _ in range(warmup_steps + extra_steps):
              qps.update(
                  predictions={
                      "DefaultTask": model_output["prediction"]
                  },
                  labels={"DefaultTask": model_output["label"]},
                  weights={"DefaultTask": model_output["weight"]},
              )
          self.assertEquals(
              qps._metrics_computations[0].warmup_examples,
              batch_size * warmup_steps * (i + 1),
          )
          self.assertEquals(
              qps._metrics_computations[0].num_examples,
              batch_size * (warmup_steps + extra_steps) * (i + 1),
          )
          # Mimic trainer crashing and loading a checkpoint.
          qps._metrics_computations[0]._steps = 0
  
imports: import unittest
  from functools import partial, update_wrapper
  from typing import Callable, Dict, List, Optional, Tuple, Type
  import torch
  import torch.distributed
  from torchrec.metrics.metrics_config import DefaultTaskInfo
  from torchrec.metrics.model_utils import parse_task_model_outputs
  from torchrec.metrics.rec_metric import RecComputeMode, RecMetric, RecTaskInfo
  from torchrec.metrics.test_utils import gen_test_batch, gen_test_tasks, metric_test_helper, rec_metric_value_test_launcher, TestMetric
  from torchrec.metrics.tower_qps import TowerQPSMetric
label_desc: Use image node_id 6 for calling the TowerQPSMetricTest obj's underlying member method code with example usage: obj.test_warmup_checkpointing() without return types