docstr: Test multi device batch normalization at the net level. This is done
  by verifying that the final batch normalization outputs and the
  gradient outputs from multiple devices are the same as those produced
  from a single device
docstr_tok: ["Test","multi","device","batch","normalization","at","the","net","level",".","This","is","done","by","verifying","that","the","final","batch","normalization","outputs","and","the","gradient","outputs","from","multiple","devices","are","the","same","as","those","produced","from","a","single","device"]
code: def _test_multi_device_bn_net_lvl(
      self, device_type, seed, batch_size
  ):
      """
      Test multi device batch normalization at the net level. This is done
      by verifying that the final batch normalization outputs and the
      gradient outputs from multiple devices are the same as those produced
      from a single device
      """
  
      # Verify that the gradients calculated over multiple devices are the
      # same as the gradients calculated over one device. These values should
      # be equivalent because combine_spatial_bn sums values over all devices
      def _verify_bn_outputs(
          devices,
          device_type,
          tolerance,
          single_device_bn_out,
          two_device_bn_out_vals,
          single_device_grads,
          two_device_grads,
      ):
          two_device_bn_out = np.concatenate(two_device_bn_out_vals)
          self.assertTrue(
              np.isclose(
                  [single_device_bn_out],
                  [two_device_bn_out],
                  atol=tolerance,
              ).all()
          )
  
          # Scalar and Bias gradients should be the same across devices
          gradient_names = ["bn_out_s_grad", "bn_out_b_grad"]
          for name in gradient_names:
              expected_grad = single_device_grads[name]
              for device in devices:
                  actual_grad = two_device_grads[device][name]
                  self.assertTrue(
                      np.isclose(
                          [actual_grad], [expected_grad], atol=tolerance
                      )
                  )
  
          # Expected tanh_grad should be the combined tanh_grad vectors
          # across the devices
          first_grad = two_device_grads[0]["tanh_grad"]
          second_grad = two_device_grads[1]["tanh_grad"]
          actual_grad = np.concatenate([first_grad, second_grad])
          expected_grad = single_device_grads["tanh_grad"]
          rel_error = np.linalg.norm(
              actual_grad - expected_grad
          ) / np.linalg.norm(expected_grad)
          self.assertTrue(rel_error < 1e-3)
  
      def _create_model(multiple_devices):
          def add_input_ops_no_combine(model):
              workspace.FeedBlob("{}_0/data".format(device_type), data)
  
          def add_input_ops_combine(model):
              half = int(batch_size / 2)
              workspace.FeedBlob(
                  "{}_0/data".format(device_type), data[:half]
              )
              workspace.FeedBlob(
                  "{}_1/data".format(device_type), data[half:]
              )
  
          def add_model_ops(model, loss_scale):
              if device_type == "gpu":
                  model.CopyCPUToGPU("data", "device_data")
                  model.Tanh("device_data", "tanh")
              else:
                  model.Tanh("data", "tanh")
              model.SpatialBN(
                  "tanh", "bn_out", 1, epsilon=epsilon, is_test=False
              )
              model.Sqr("bn_out", "sqr")
              loss = model.SumElements("sqr", "loss")
              return [loss]
  
          def add_optimizer(model):
              return optimizer.build_sgd(model, 0.1)
  
          if multiple_devices:
              input_fun = add_input_ops_combine
              devices = [0, 1]
              combine_spatial_bn = True
          else:
              input_fun = add_input_ops_no_combine
              devices = [0]
              combine_spatial_bn = False
          model = cnn.CNNModelHelper(order="NCHW", name="test")
          data_parallel_model.Parallelize(
              model,
              input_builder_fun=input_fun,
              forward_pass_builder_fun=add_model_ops,
              optimizer_builder_fun=add_optimizer,
              devices=devices,
              cpu_device=device_type == "cpu",
              shared_model=False,
              combine_spatial_bn=combine_spatial_bn,
          )
          return model
  
      devices = [0, 1]
      epsilon = 1e-3
      tolerance = 1e-3
      # We are generating random data
      np.random.seed(seed)
      data = np.random.rand(batch_size, 1, 1, 1).astype(np.float32)
      data = np.reshape(data, (batch_size, 1, 1, 1))
  
      # Get values calculated without combine_spatial_bn
      workspace.ResetWorkspace()
      model_no_combine = _create_model(multiple_devices=False)
      workspace.RunNetOnce(model_no_combine.param_init_net)
      workspace.RunNetOnce(model_no_combine.net)
      single_device_bn_out = workspace.FetchBlob(
          "{}_0/bn_out".format(device_type)
      )
      single_device_grads = {}
      single_device_grads["bn_out_s_grad"] = workspace.FetchBlob(
          "{}_0/bn_out_s_grad".format(device_type)
      )
      single_device_grads["bn_out_b_grad"] = workspace.FetchBlob(
          "{}_0/bn_out_b_grad".format(device_type)
      )
      single_device_grads["tanh_grad"] = workspace.FetchBlob(
          "{}_0/tanh_grad".format(device_type)
      )
  
      # Get values calculated over multiple devices with combine_spatial_bn true
      workspace.ResetWorkspace()
      model_combine = _create_model(multiple_devices=True)
      workspace.RunNetOnce(model_combine.param_init_net)
      workspace.RunNetOnce(model_combine.net)
      two_device_bn_out_vals = []
      two_device_grads = {}
      for device in devices:
          bn_out_blob = "{}_{}/bn_out".format(device_type, device)
          two_device_bn_out_vals.append(
              workspace.FetchBlob(bn_out_blob)
          )
          two_device_grads[device] = {}
          two_device_grads[device][
              "bn_out_s_grad"
          ] = workspace.FetchBlob(
              "{}_{}/bn_out_s_grad".format(device_type, device)
          )
          two_device_grads[device][
              "bn_out_b_grad"
          ] = workspace.FetchBlob(
              "{}_{}/bn_out_b_grad".format(device_type, device)
          )
          two_device_grads[device]["tanh_grad"] = workspace.FetchBlob(
              "{}_{}/tanh_grad".format(device_type, device)
          )
  
      # Check to see if the combined values are equivalent
      _verify_bn_outputs(
          devices,
          device_type,
          tolerance,
          single_device_bn_out,
          two_device_bn_out_vals,
          single_device_grads,
          two_device_grads,
      )
  
imports: from multiprocessing import Process, Queue
  import numpy
  import os
  import shutil
  import tempfile
  import unittest
  import time
  from unittest.mock import Mock
  from hypothesis import assume, given, settings
  import hypothesis.strategies
  from caffe2.proto import caffe2_pb2
  from caffe2.python import brew, core, cnn, data_parallel_model, dyndep, model_helper, optimizer, rnn_cell, workspace
  from caffe2.python.test_util import TestCase
label_desc: Use image node_id 15 for calling the DataParallelModelTest obj's underlying member method code with example usage: obj._test_multi_device_bn_net_lvl(device_type, seed, batch_size) and returns: model, list, optimizer