class_docstr: Data-parallel training in "replicated" mode,
  where each GPU contains a replicate of the whole model.
  It will build one tower on each GPU under its own variable scope.
  Each gradient update is averaged or summed across or GPUs through NCCL.
  
  It is an equivalent of ``--variable_update=replicated`` in
  `tensorflow/benchmarks <https://github.com/tensorflow/benchmarks>`_.
class_docstr_tok: ["Data-parallel","training","in","``","replicated","''","mode",",","where","each","GPU","contains","a","replicate","of","the","whole","model",".","It","will","build","one","tower","on","each","GPU","under","its","own","variable","scope",".","Each","gradient","update","is","averaged","or","summed","across","or","GPUs","through","NCCL",".","It","is","an","equivalent","of","``","--","variable_update=replicated","``","in","`","tensorflow\/benchmarks","<","https",":","\/\/github.com\/tensorflow\/benchmarks",">","`","_","."]
docstr: Reduce the gradients, apply them with the optimizer,
  and set self.grads to #GPU number of lists of (g, v), containing the all-reduced gradients on each device.
  
  Args:
      grad_list ([[(grad, var), ...], ...]): #GPU lists to be reduced. Each is the gradients computed on each GPU.
      get_opt_fn (-> tf.train.Optimizer): callable which returns an optimizer
  
  Returns:
      (tf.Operation, tf.Operation)
  
      1. the training op.
  
      2. the op which sync variables from GPU 0 to other GPUs.
          It has to be run before the training has started.
          And you can optionally run it later to sync non-trainable variables.
docstr_tok: ["Reduce","the","gradients",",","apply","them","with","the","optimizer",",","and","set","self.grads","to","#","GPU","number","of","lists","of","(","g",",","v",")",",","containing","the","all-reduced","gradients","on","each","device",".","Args",":","grad_list","(","[","[","(","grad",",","var",")",",","...","]",",","...","]",")",":","#","GPU","lists","to","be","reduced",".","Each","is","the","gradients","computed","on","each","GPU",".","get_opt_fn","(","-",">","tf.train.Optimizer",")",":","callable","which","returns","an","optimizer","Returns",":","(","tf.Operation",",","tf.Operation",")","1.","the","training","op",".","2.","the","op","which","sync","variables","from","GPU","0","to","other","GPUs",".","It","has","to","be","run","before","the","training","has","started",".","And","you","can","optionally","run","it","later","to","sync","non-trainable","variables","."]
code: def build(self, grad_list, get_opt_fn):
      """
      Reduce the gradients, apply them with the optimizer,
      and set self.grads to #GPU number of lists of (g, v), containing the all-reduced gradients on each device.
  
      Args:
          grad_list ([[(grad, var), ...], ...]): #GPU lists to be reduced. Each is the gradients computed on each GPU.
          get_opt_fn (-> tf.train.Optimizer): callable which returns an optimizer
  
      Returns:
          (tf.Operation, tf.Operation)
  
          1. the training op.
  
          2. the op which sync variables from GPU 0 to other GPUs.
              It has to be run before the training has started.
              And you can optionally run it later to sync non-trainable variables.
      """
      assert len(grad_list) == len(self.towers)
      raw_devices = ["/gpu:{}".format(k) for k in self.towers]
  
      DataParallelBuilder._check_grad_list(grad_list)
  
      dtypes = {x[0].dtype.base_dtype for x in grad_list[0]}
      dtypes_nccl_supported = [tf.float32, tf.float64]
      if get_tf_version_tuple() >= (1, 8):
          dtypes_nccl_supported.append(tf.float16)
      valid_for_nccl = all(k in dtypes_nccl_supported for k in dtypes)
      if self._mode == "nccl" and not valid_for_nccl:
          logger.warn(
              "Cannot use mode='nccl' because some gradients have unsupported types. Fallback to mode='cpu'"
          )
          self._mode = "gpu"
  
      all_grads, all_vars = split_grad_list(grad_list)
  
      def do_allreduce(all_grads):
          # use allreduce from tf-benchmarks
          # from .batch_allreduce import AllReduceSpecAlgorithm
          # algo = AllReduceSpecAlgorithm('nccl', list(range(8)), 0, 10)
          # all_grads, warmup_ops = algo.batch_all_reduce(all_grads, 1, True, False)
          # print("WARMUP OPS", warmup_ops)
          if self._mode in ["nccl", "collective"]:
              # #gpu x #param
              all_grads = allreduce_grads(
                  all_grads, average=self._average, mode=self._mode
              )
          elif self._mode == "hierarchical":
              all_grads = allreduce_grads_hierarchical(
                  all_grads, raw_devices, average=self._average
              )
          else:
              devices = (
                  ["/cpu:0"] if self._mode == "cpu" else raw_devices
              )
              all_grads = allreduce_grads_naive(
                  all_grads, devices=devices, average=self._average
              )
              all_grads = [all_grads] * len(self.towers)
          return all_grads
  
      use_packer = self._mode in ["hierarchical"]
      if use_packer:
          packer = GradientPacker(len(raw_devices))
          use_packer = packer.compute_strategy(
              all_grads[0]
          )  # may fail to pack
      if use_packer:
          all_grads = packer.pack_all(all_grads, raw_devices)
      all_grads = do_allreduce(all_grads)  # all the work happens here
      if use_packer:
          all_grads = packer.unpack_all(all_grads, raw_devices)
  
      self.grads = merge_grad_list(all_grads, all_vars)
  
      train_ops = []
      opt = get_opt_fn()
      with tf.name_scope("apply_gradients"):
          for idx, grad_and_vars in enumerate(self.grads):
              with tf.device(raw_devices[idx]):
                  # apply_gradients may create variables. Make them LOCAL_VARIABLES
                  with override_to_local_variable(enable=idx > 0):
                      train_ops.append(
                          opt.apply_gradients(
                              grad_and_vars,
                              name="apply_grad_{}".format(idx),
                          )
                      )
      train_op = tf.group(*train_ops, name="train_op")
  
      if len(self.towers) > 1:
          with tf.name_scope("sync_variables"):
              post_init_op = (
                  SyncMultiGPUReplicatedBuilder.get_post_init_ops()
              )
      else:
          post_init_op = None
      return train_op, post_init_op
  
imports: import copy
  import pprint
  import re
  from abc import ABCMeta, abstractmethod
  from contextlib import contextmanager
  import six
  import tensorflow
  from ..compat import tfv1
  from ..tfutils.common import get_tf_version_tuple
  from ..tfutils.gradproc import ScaleGradient
  from ..tfutils.tower import TrainTowerContext
  from ..utils import logger
  from ..utils.develop import HIDE_DOC
  from .utils import GradientPacker, LeastLoadedDeviceSetter, aggregate_grads_colocate, allreduce_grads_naive, allreduce_grads, allreduce_grads_hierarchical, merge_grad_list, override_to_local_variable, split_grad_list
label_desc: Use image node_id 3 for calling the SyncMultiGPUReplicatedBuilder obj's underlying member method code with example usage: obj.build(grad_list, get_opt_fn) and returns: train_op,  post_init_op, all_grads