code: def _test_throughput(
      self, time_mock: Mock, warmup_steps: int
  ) -> None:
      update_timestamps = [
          10,
          11,
          12,
          14,
          15,
          17,
          18,
          20,
          21,
          22,
          25,
          29,
          30,
      ]
      update_timestamps.sort()
      throughput_metric = ThroughputMetric(
          batch_size=self.batch_size,
          world_size=self.world_size,
          window_seconds=10,
          warmup_steps=warmup_steps,
      )
  
      window_time_lapse_buffer = []
      window_time_lapse = 0
      for i in range(len(update_timestamps)):
          time_mock.return_value = update_timestamps[i]
          throughput_metric.update()
          if i >= warmup_steps:
              window_time_lapse_buffer.append(
                  update_timestamps[i] - update_timestamps[i - 1]
              )
              window_time_lapse += (
                  update_timestamps[i] - update_timestamps[i - 1]
              )
          ret = throughput_metric.compute()
  
          total_examples = self.world_size * self.batch_size * (i + 1)
          if i < warmup_steps:
              self.assertEqual(
                  ret,
                  {
                      "throughput-throughput|total_examples": total_examples
                  },
              )
              continue
  
          lifetime_examples = total_examples - (
              self.world_size * self.batch_size * warmup_steps
          )
          lifetime_throughput = lifetime_examples / (
              update_timestamps[i] - update_timestamps[warmup_steps - 1]
          )
  
          while window_time_lapse > 10:
              window_time_lapse -= window_time_lapse_buffer.pop(0)
          window_throughput = (
              len(window_time_lapse_buffer)
              * self.world_size
              * self.batch_size
              / window_time_lapse
          )
  
          self.assertEqual(
              ret["throughput-throughput|lifetime_throughput"],
              lifetime_throughput,
          )
          self.assertEqual(
              ret["throughput-throughput|window_throughput"],
              window_throughput,
          )
          self.assertEqual(
              ret["throughput-throughput|total_examples"],
              total_examples,
          )
  
file_path: torchrec/torchrec/metrics/tests/test_throughput.py
imports: import unittest
  from unittest.mock import Mock, patch
  from torchrec.metrics.throughput import ThroughputMetric
label_desc: Use image node_id 4 for calling the ThroughputMetricTest obj's underlying member method code with example usage: obj._test_throughput(time_mock, warmup_steps) without return types