code: """
  A simple tool to compare the performance of different impls of
  DistributedDataParallel on resnet50, three flavors:
  
  1. DistributedDataParallel, which has a python wrapper and C++ core to do
     gradient distribution and reduction. It's current production version.
  
  2. PythonDDP with async gradient reduction.
  
  3. PythonDDP with synchrous gradient reduction.
  
  Example::
      >>> modify configs in main func
      >>> python compare_ddp.py
      >>> Sample out: compare_ddp_sample.md
  """
  
  import numpy as np
  import os
  import pickle
  import glob
  import python_ddp
  import torch
  import torch.distributed as dist
  import torch.multiprocessing as mp
  import torch.nn as nn
  import torch.optim as optim
  import torchvision.models as models
  
  from collections import OrderedDict
  from enum import Enum
  from tabulate import tabulate
  from torch.nn.parallel import DistributedDataParallel as DDP
  
  class DDPOption(Enum):
      DDP_CPP_CORE = 1
      PYTHON_DDP_SYNC_REDUCTION = 2
      PYTHON_DDP_ASYNC_REDUCTION = 3
  
  class LatencyData:
      __slots__ = ["buffer_size_in_M", "ddp_option", "rank", "metrics"]
  
      def __init__(self, buffer_size_in_M, ddp_option, rank, metrics):
          self.buffer_size_in_M = buffer_size_in_M
          self.ddp_option = ddp_option
          self.rank = rank
          self.metrics = metrics
  
  def serialize(buffer_size_in_M, ddp_option, rank, metrics,
                data_dir="./tmp", ext="ddpraw"):
      if not os.path.exists(data_dir):
          print(f'{data_dir} not exist, mkdir {data_dir}')
          os.mkdir(data_dir)
      file_name = "buffer_size_{}M_rank{}_{}.{}".format(
          buffer_size_in_M, rank, ddp_option, ext)
      file_path = os.path.join(data_dir, file_name)
      print("Writing metrics to file: '{}'".format(file_path))
      data = LatencyData(buffer_size_in_M, ddp_option, rank, metrics)
      with open(file_path, "wb") as f:
          pickle.dump(data, f, pickle.HIGHEST_PROTOCOL)
      print(f"Wrote metrics to '{file_path}''")
  
  def load_detailed_metrics(data_dir="./tmp", ext="ddpraw"):
      assert os.path.exists(data_dir)
      file_pattern = os.path.join(data_dir, f"*.{ext}")
      files = glob.glob(file_pattern)
      print("load_detailed_metrics found {} files".format(len(files)))
      buffer_size_to_metrics = OrderedDict()
      for file_path in files:
          with open(file_path, "rb") as f:
              data = pickle.load(f)
          # Add data to buffer_size_to_metrics
          buffer_size = data.buffer_size_in_M
          if buffer_size not in buffer_size_to_metrics:
              buffer_size_to_metrics[buffer_size] = {}
          metrics = buffer_size_to_metrics.get(buffer_size)
          assert metrics is not None
          metrics[data.ddp_option] = data.metrics
      return buffer_size_to_metrics
  
  def setup(rank, world_size):
      os.environ['MASTER_ADDR'] = 'localhost'
      os.environ['MASTER_PORT'] = '12355'
  
      # initialize the process group
      dist.init_process_group("gloo", rank=rank, world_size=world_size)
  
  def create_ddp_model(module, rank, pg, ddp_option, buffer_size_in_M):
      """Helper to create DDPModel. """
      if ddp_option == DDPOption.DDP_CPP_CORE:
          ddp_model = DDP(module, device_ids=[rank],
                          process_group=pg,
                          bucket_cap_mb=buffer_size_in_M)
          ddp_model._set_static_graph()
          return ddp_model
      elif ddp_option == DDPOption.PYTHON_DDP_SYNC_REDUCTION:
          M = 2 ** 20
          return python_ddp.PythonDDP(module, pg, False, buffer_size=buffer_size_in_M * M)
      elif ddp_option == DDPOption.PYTHON_DDP_ASYNC_REDUCTION:
          M = 2 ** 20
          return python_ddp.PythonDDP(module, pg, True, buffer_size=buffer_size_in_M * M)
      else:
          raise NotImplementedError
  
  def run_ddp(rank, world_size, epochs, ddp_option, buffer_size_in_M, warmup_iterations=20):
      print(f'Invoked run_ddp rank {rank}')
      assert epochs > warmup_iterations
  
      # Setup
      print("setting up ... ")
      setup(rank, world_size)
      torch.manual_seed(rank)
      torch.cuda.manual_seed(rank)
      device = torch.device('cuda:%d' % rank)
      print('setup done')
  
      # Create ResNet50 module and wrap in DDP module.
      pg = dist.distributed_c10d._get_default_group()
      model = models.resnet50().to(device)
      ddp_model = create_ddp_model(model, rank, pg, ddp_option, buffer_size_in_M)
      assert ddp_model is not None
  
      loss_fn = nn.MSELoss()
      optimizer = optim.SGD(ddp_model.parameters(), lr=0.001)
  
      # Container to hold: event -> list of events in milliseconds
      MODEL_FORWARD = "forward"
      MODEL_BACKWARD = "backward"
      metrics = {MODEL_FORWARD: [], MODEL_BACKWARD: []}
  
      for epoch in range(epochs):
          if epoch % 10 == 0:
              print(f'Epoch {epoch}/{epochs} ...')
  
          start = torch.cuda.Event(enable_timing=True)
          end = torch.cuda.Event(enable_timing=True)
  
          # TODO(bowangbj): Switch to real training set from ImageNet.
          inputs = torch.rand([32, 3, 224, 224], device=device)
          labels = torch.rand([32, 1000], device=device)
  
          # Forward
          start.record()
          outputs = ddp_model(inputs)
          loss = loss_fn(outputs, labels)
  
          end.record()
          torch.cuda.synchronize()
          if epoch >= warmup_iterations:
              metrics[MODEL_FORWARD].append(start.elapsed_time(end))
  
          # Backward
          start.record()
          loss.backward()
          # Reduce all grad, this is needed for non-DDP_CPP_CORE since the hook
          # for all_reduce does not exist yet.
          if ddp_option != DDPOption.DDP_CPP_CORE:
              ddp_model.all_reduce_grads()
          end.record()
          torch.cuda.synchronize()
          if epoch >= warmup_iterations:
              metrics[MODEL_BACKWARD].append(start.elapsed_time(end))
  
          # Optimization
          optimizer.step()
          optimizer.zero_grad()
  
      if rank == 0:
          print(f"\nMetrics for GPU {rank}, ddp_option={ddp_option}, buffer_size={buffer_size_in_M}M")
          print(f"Skipped {warmup_iterations} CUDA warmpup iterations. ")
          for step, elapsed_milliseconds in metrics.items():
              A = np.array(elapsed_milliseconds)
              print(' {N} iterations, {step}, mean={mean} ms, median={median} ms, p90={p90} ms, p99={p99} ms'.format(
                  N=len(A), step=step, mean=np.mean(A),
                  median=np.percentile(A, 50), p90=np.percentile(A, 90),
                  p99=np.percentile(A, 99)))
  
          # Serialize the raw data to be used to compute summary. Didn't choose to
          # maintain a global object holding the metrics b/c mp.spawn tries to
          # fork all the arguments before spawning new process thus it's infeasible
          # save global states in an object.
          serialize(buffer_size_in_M, ddp_option, rank, metrics)
  
  def append_delta(row_list, base, exp):
      percent = 100 * ((exp - base) / base)
      row_list.append(percent)
  
  def print_summary(buffer_size_to_metrics):
      # metrics: {ddp_option, Metrics}
      # Metrics: step -> [latency]
      for buffer_size, metrics in buffer_size_to_metrics.items():
          assert DDPOption.DDP_CPP_CORE in metrics.keys()
          baseline = metrics.get(DDPOption.DDP_CPP_CORE)
          print(f"=== Summary for buffer_size: {buffer_size}M === ")
          for step in baseline.keys():
              # step takes value from [forward, backward]
              # compute latency for each step into a table, each row is looks like
              # [option, mean, diff, mean, diff, p90, diff, p95, diff, p99, diff]
              data = []
              baseline_latencies = baseline.get(step)
              assert baseline_latencies is not None
              A_baseline = np.array(baseline_latencies)
              for ddp_option, exp_metrics in metrics.items():
                  exp_latencies = exp_metrics.get(step)
                  assert exp_latencies is not None
                  A_exp = np.array(exp_latencies)
                  # Yield option, mean, p50, p90, p95, p99 and delta.
                  row = [ddp_option]
                  row.append(np.mean(A_exp))
                  append_delta(row, np.mean(A_baseline), np.mean(A_exp))
                  for px in [50, 90, 95, 99]:
                      base = np.percentile(A_baseline, px)
                      exp = np.percentile(A_exp, px)
                      row.append(exp)
                      append_delta(row, base, exp)
                  data.append(row)
  
              # Output buffer_size, step as a table.
              print(tabulate(data,
                             headers=[f"DDP: [{step}]", "Mean", "delta%",
                                      "mean", "delta%", "p90", "delta%",
                                      "p95", "delta%%", "p99", "delta%"]))
              print("\n")
  
  def main():
      world_size = 2
      epochs = 120
  
      # resnet50 model facts:
      # total_param_count = 161
      # total_elements = 25557032 ~= 24.37M
      # param_max_elements = 2359296 ~= 2.25M
      # Try different bucket sizes.
      buffer_size_in_mbs = [3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27]
      print("buffer_size_in_mbs: " + str(buffer_size_in_mbs))
      for buffer_size_in_M in buffer_size_in_mbs:
          print("\n\n=== NEW EXPERIMENT: buffer_size={}M, {} epochs, world_size={} ===".format(
              buffer_size_in_M, epochs, world_size))
          options = [
              DDPOption.DDP_CPP_CORE,
              DDPOption.PYTHON_DDP_ASYNC_REDUCTION,
              DDPOption.PYTHON_DDP_SYNC_REDUCTION
          ]
          for option in options:
              print("Measuring option: {} ... ".format(option))
              mp.spawn(run_ddp,
                       args=(world_size, epochs, option, buffer_size_in_M),
                       nprocs=world_size,
                       join=True)
  
      print("\n Generating summaries ... ")
      buffer_size_to_metrics = load_detailed_metrics(data_dir="./tmp", ext="ddpraw")
      print_summary(buffer_size_to_metrics)
  
  if __name__ == "__main__" :
      main()
   
  
  import collections
  
  import caffe2.python.hypothesis_test_util as hu
  import hypothesis.strategies as st
  import numpy as np
  from caffe2.python import core, dyndep, workspace
  from hypothesis import given, settings
  
  
  dyndep.InitOpsLibrary("//caffe2/caffe2/quantization/server:dnnlowp_ops")
  workspace.GlobalInit(["caffe2", "--caffe2_omp_num_threads=11"])
  
  
  class DNNLowPTanhOpTest(hu.HypothesisTestCase):
      @given(size=st.integers(1024, 2048), is_empty=st.booleans(), **hu.gcs_cpu_only)
      @settings(max_examples=10, deadline=None)
      def test_dnnlowp_tanh(self, size, is_empty, gc, dc):
          if is_empty:
              size = 0
  
          X = (np.random.rand(size) * 10 - 5).astype(np.float32)
  
          Output = collections.namedtuple("Output", ["Y", "op_type", "engine"])
          outputs = []
  
          op_engine_list = [("Tanh", ""), ("Tanh", "DNNLOWP"), ("Int8Tanh", "DNNLOWP")]
  
          for op_type, engine in op_engine_list:
              net = core.Net("test_net")
  
              if engine == "DNNLOWP":
                  quantize = core.CreateOperator(
                      "Quantize",
                      ["X"],
                      ["X_q"],
                      engine=engine,
                      device_option=gc,
                      followed_by="Tanh",
                  )
                  net.Proto().op.extend([quantize])
  
              tanh = core.CreateOperator(
                  op_type,
                  ["X_q" if engine == "DNNLOWP" else "X"],
                  ["Y_q" if engine == "DNNLOWP" else "Y"],
                  engine=engine,
                  device_option=gc,
              )
              net.Proto().op.extend([tanh])
  
              if engine == "DNNLOWP":
                  dequantize = core.CreateOperator(
                      "Dequantize", ["Y_q"], ["Y"], engine=engine, device_option=gc
                  )
                  net.Proto().op.extend([dequantize])
  
              self.ws.create_blob("X").feed(X, device_option=gc)
              self.ws.run(net)
              outputs.append(
                  Output(Y=self.ws.blobs["Y"].fetch(), op_type=op_type, engine=engine)
              )
  
          for o in outputs:
              np.testing.assert_allclose(o.Y, outputs[0].Y, atol=0.02, rtol=0)
   #!/usr/bin/env python3
  
  
  
  
  
  
  from hypothesis import given, settings
  import hypothesis.strategies as st
  from multiprocessing import Process
  
  import numpy as np
  import tempfile
  import shutil
  
  import caffe2.python.hypothesis_test_util as hu
  import unittest
  
  op_engine = 'GLOO'
  
  class TemporaryDirectory:
      def __enter__(self):
          self.tmpdir = tempfile.mkdtemp()
          return self.tmpdir
  
      def __exit__(self, type, value, traceback):
          shutil.rmtree(self.tmpdir)
  
  
  def allcompare_process(filestore_dir, process_id, data, num_procs):
      from caffe2.python import core, data_parallel_model, workspace, lazy_dyndep
      from caffe2.python.model_helper import ModelHelper
      from caffe2.proto import caffe2_pb2
      lazy_dyndep.RegisterOpsLibrary("@/caffe2/caffe2/distributed:file_store_handler_ops")
  
      workspace.RunOperatorOnce(
          core.CreateOperator(
              "FileStoreHandlerCreate", [], ["store_handler"], path=filestore_dir
          )
      )
      rendezvous = dict(
          kv_handler="store_handler",
          shard_id=process_id,
          num_shards=num_procs,
          engine=op_engine,
          exit_nets=None
      )
  
      model = ModelHelper()
      model._rendezvous = rendezvous
  
      workspace.FeedBlob("test_data", data)
  
      data_parallel_model._RunComparison(
          model, "test_data", core.DeviceOption(caffe2_pb2.CPU, 0)
      )
  
  
  class TestLazyDynDepAllCompare(hu.HypothesisTestCase):
      @given(
          d=st.integers(1, 5), n=st.integers(2, 11), num_procs=st.integers(1, 8)
      )
      @settings(deadline=None)
      def test_allcompare(self, d, n, num_procs):
          dims = []
          for _ in range(d):
              dims.append(np.random.randint(1, high=n))
          test_data = np.random.ranf(size=tuple(dims)).astype(np.float32)
  
          with TemporaryDirectory() as tempdir:
              processes = []
              for idx in range(num_procs):
                  process = Process(
                      target=allcompare_process,
                      args=(tempdir, idx, test_data, num_procs)
                  )
                  processes.append(process)
                  process.start()
  
              while len(processes) > 0:
                  process = processes.pop()
                  process.join()
  
  class TestLazyDynDepError(unittest.TestCase):
      def test_errorhandler(self):
          from caffe2.python import core, lazy_dyndep
          import tempfile
  
          with tempfile.NamedTemporaryFile() as f:
              lazy_dyndep.RegisterOpsLibrary(f.name)
  
              def handler(e):
                  raise ValueError("test")
              lazy_dyndep.SetErrorHandler(handler)
              with self.assertRaises(ValueError, msg="test"):
                  core.RefreshRegisteredOperators()
  
      def test_importaftererror(self):
          from caffe2.python import core, lazy_dyndep
          import tempfile
  
          with tempfile.NamedTemporaryFile() as f:
              lazy_dyndep.RegisterOpsLibrary(f.name)
  
              def handler(e):
                  raise ValueError("test")
              lazy_dyndep.SetErrorHandler(handler)
              with self.assertRaises(ValueError):
                  core.RefreshRegisteredOperators()
  
              def handlernoop(e):
                  raise
              lazy_dyndep.SetErrorHandler(handlernoop)
              lazy_dyndep.RegisterOpsLibrary("@/caffe2/caffe2/distributed:file_store_handler_ops")
              core.RefreshRegisteredOperators()
  
      def test_workspacecreatenet(self):
          from caffe2.python import workspace, lazy_dyndep
          import tempfile
  
          with tempfile.NamedTemporaryFile() as f:
              lazy_dyndep.RegisterOpsLibrary(f.name)
              called = False
  
              def handler(e):
                  raise ValueError("test")
              lazy_dyndep.SetErrorHandler(handler)
              with self.assertRaises(ValueError, msg="test"):
                  workspace.CreateNet("fake")
  
  
  if __name__ == "__main__":
      unittest.main()
   import os
  import tempfile
  
  from torch.testing._internal.common_utils import TestCase, run_tests
  import tools.codegen.gen_backend_stubs
  
  path = os.path.dirname(os.path.realpath(__file__))
  gen_backend_stubs_path = os.path.join(path, '../tools/codegen/gen_backend_stubs.py')
  
  # gen_backend_stubs.py is an integration point that is called directly by external backends.
  # The tests here are to confirm that badly formed inputs result in reasonable error messages.
  class TestGenBackendStubs(TestCase):
  
      def assert_success_from_gen_backend_stubs(self, yaml_str: str) -> str:
          with tempfile.NamedTemporaryFile(mode='w') as fp:
              fp.write(yaml_str)
              fp.flush()
              tools.codegen.gen_backend_stubs.run(fp.name, '', True)
  
      def get_errors_from_gen_backend_stubs(self, yaml_str: str) -> str:
          with tempfile.NamedTemporaryFile(mode='w') as fp:
              fp.write(yaml_str)
              fp.flush()
              try:
                  tools.codegen.gen_backend_stubs.run(fp.name, '', True)
              except AssertionError as e:
                  # Scrub out the temp file name from any error messages to simplify assertions.
                  return str(e).replace(fp.name, '')
              self.fail('Expected gen_backend_stubs to raise an AssertionError, but it did not.')
  
      def test_valid_single_op(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - abs'''
          self.assert_success_from_gen_backend_stubs(yaml_str)
  
      def test_valid_multiple_ops(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - add.Tensor
  - abs'''
          self.assert_success_from_gen_backend_stubs(yaml_str)
  
      def test_valid_zero_ops(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:'''
          self.assert_success_from_gen_backend_stubs(yaml_str)
  
      def test_valid_zero_ops_doesnt_require_backend_dispatch_key(self):
          yaml_str = '''\
  backend: BAD_XLA
  cpp_namespace: torch_xla
  supported:'''
          # External codegen on a yaml file with no operators is effectively a no-op,
          # so there's no reason to parse the backend
          self.assert_success_from_gen_backend_stubs(yaml_str)
  
      def test_valid_with_autograd_ops(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - abs
  autograd:
  - add.Tensor'''
          # External codegen on a yaml file with no operators is effectively a no-op,
          # so there's no reason to parse the backend
          self.assert_success_from_gen_backend_stubs(yaml_str)
  
      def test_missing_backend(self):
          yaml_str = '''\
  cpp_namespace: torch_xla
  supported:
  - abs'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''You must provide a value for "backend"''')
  
      def test_empty_backend(self):
          yaml_str = '''\
  backend:
  cpp_namespace: torch_xla
  supported:
  - abs'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''You must provide a value for "backend"''')
  
      def test_backend_invalid_dispatch_key(self):
          yaml_str = '''\
  backend: NOT_XLA
  cpp_namespace: torch_xla
  supported:
  - abs'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''\
  unknown dispatch key NOT_XLA
    The provided value for "backend" must be a valid DispatchKey, but got NOT_XLA.''')  # noqa: B950
  
      def test_missing_cpp_namespace(self):
          yaml_str = '''\
  backend: XLA
  supported:
  - abs'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''You must provide a value for "cpp_namespace"''')
  
      def test_whitespace_cpp_namespace(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace:\t
  supported:
  - abs'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''You must provide a value for "cpp_namespace"''')
  
      # supported is a single item (it should be a list)
      def test_nonlist_supported(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported: abs'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''expected "supported" to be a list, but got: abs (of type <class 'str'>)''')
  
      # supported contains an op that isn't in native_functions.yaml
      def test_supported_invalid_op(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - abs_BAD'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''Found an invalid operator name: abs_BAD''')
  
      # The backend is valid, but doesn't have a valid autograd key. They can't override autograd kernels in that case.
      # Only using Vulkan here because it has a valid backend key but not an autograd key- if this changes we can update the test.
      def test_backend_has_no_autograd_key_but_provides_entries(self):
          yaml_str = '''\
  backend: Vulkan
  cpp_namespace: torch_vulkan
  supported:
  - add
  autograd:
  - sub'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''Found an invalid operator name: add''')  # noqa: B950
  
      # in an operator group, currently all operators must either be registered to the backend or autograd kernel.
      # Here, functional and out mismatch
      def test_backend_autograd_kernel_mismatch_out_functional(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - add.Tensor
  autograd:
  - add.out'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''Currently, all variants of an op must either be registered to a backend key, or to a backend's autograd key. They cannot be mix and matched. If this is something you need, feel free to create an issue! add is listed under "supported", but add_out is listed under "autograd".''')  # noqa: B950
  
      # in an operator group, currently all operators must either be registered to the backend or autograd kernel.
      # Here, functional and inplace mismatch
      def test_backend_autograd_kernel_mismatch_functional_inplace(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - add.Tensor
  autograd:
  - add_.Tensor'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''Currently, all variants of an op must either be registered to a backend key, or to a backend's autograd key. They cannot be mix and matched. If this is something you need, feel free to create an issue! add is listed under "supported", but add_ is listed under "autograd".''')  # noqa: B950
  
      # Currently, the same operator can't be listed under both 'supported' and 'autograd', which would
      # involve registering the same kernel to both the XLA and AutogradXLA keys.
      # If we need that functionality in the future, we'll need to augment the codegen.
      def test_op_appears_in_supported_and_autograd_lists(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - add.Tensor
  autograd:
  - add.Tensor'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, '''Currently, all variants of an op must either be registered to a backend key, or to a backend's autograd key. They cannot be mix and matched. If this is something you need, feel free to create an issue! add is listed under "supported", but add is listed under "autograd".''')  # noqa: B950
  
      # unrecognized extra yaml key
      def test_unrecognized_key(self):
          yaml_str = '''\
  backend: XLA
  cpp_namespace: torch_xla
  supported:
  - abs
  invalid_key: invalid_val'''
          output_error = self.get_errors_from_gen_backend_stubs(yaml_str)
          self.assertExpectedInline(output_error, ''' contains unexpected keys: invalid_key. Only the following keys are supported: backend, cpp_namespace, extra_headers, supported, autograd''')  # noqa: B950
  
  
  if __name__ == '__main__':
      run_tests()
  
apis: torch.distributed.distributed_c10d._get_default_group torch.nn.MSELoss torch.multiprocessing.spawn torch.cuda.manual_seed torch.distributed.init_process_group torch.manual_seed torch.rand torch.cuda.synchronize torch.cuda.Event torch.nn.parallel.DistributedDataParallel numpy.array torch.device numpy.percentile numpy.mean
apis: numpy.random.rand numpy.testing.assert_allclose
apis: numpy.random.randint
apis: torch.testing._internal.common_utils.run_tests