class_docstr: Distributed replicated training.
  Each worker process builds the same model on one or more GPUs.
  Gradients across GPUs are averaged within the worker,
  and get synchronously applied to the global copy of variables located on PS.
  Then each worker copy the latest variables from PS back to local.
  
  It is an equivalent of ``--variable_update=distributed_replicated`` in
  `tensorflow/benchmarks <https://github.com/tensorflow/benchmarks>`_.
  Note that the performance of this trainer is still not satisfactory,
  and TensorFlow team is not actively maintaining distributed training features.
  Check :class:`HorovodTrainer` and
  `ResNet-Horovod <https://github.com/tensorpack/benchmarks/tree/master/ResNet-Horovod>`_
  for better distributed training support.
  
  Note:
      1. Gradients are not averaged across workers, but applied to PS variables
         directly (either with or without locking depending on the optimizer).
      2. Some details about collections: all variables created inside tower
         will become local variables,
         and a clone will be made in global variables for all trainable/model variables.
  
  Example:
  
      .. code-block:: python
  
          # Create the server object like this:
          hosts = ['host1.com', 'host2.com']
          cluster_spec = tf.train.ClusterSpec({
              'ps': [h + ':2222' for h in hosts],
              'worker': [h + ':2223' for h in hosts]
          })
          server = tf.train.Server(
              cluster_spec, job_name=args.job, task_index=args.task,
              config=get_default_sess_config())
          # initialize trainer with this server object
  
      .. code-block:: none
  
          # Start training like this:
          (host1)$ ./train.py --job worker --task 0
          (host1)$ CUDA_VISIBLE_DEVICES= ./train.py --job ps --task 0
          (host2)$ ./train.py --job worker --task 1
          (host2)$ CUDA_VISIBLE_DEVICES= ./train.py --job ps --task 1
class_docstr_tok: ["Distributed","replicated","training",".","Each","worker","process","builds","the","same","model","on","one","or","more","GPUs",".","Gradients","across","GPUs","are","averaged","within","the","worker",",","and","get","synchronously","applied","to","the","global","copy","of","variables","located","on","PS",".","Then","each","worker","copy","the","latest","variables","from","PS","back","to","local",".","It","is","an","equivalent","of","``","--","variable_update=distributed_replicated","``","in","`","tensorflow\/benchmarks","<","https",":","\/\/github.com\/tensorflow\/benchmarks",">","`","_",".","Note","that","the","performance","of","this","trainer","is","still","not","satisfactory",",","and","TensorFlow","team","is","not","actively","maintaining","distributed","training","features",".","Check",":","class",":","`","HorovodTrainer","`","and","`","ResNet-Horovod","<","https",":","\/\/github.com\/tensorpack\/benchmarks\/tree\/master\/ResNet-Horovod",">","`","_","for","better","distributed","training","support",".","Note",":","1",".","Gradients","are","not","averaged","across","workers",",","but","applied","to","PS","variables","directly","(","either","with","or","without","locking","depending","on","the","optimizer",")",".","2",".","Some","details","about","collections",":","all","variables","created","inside","tower","will","become","local","variables",",","and","a","clone","will","be","made","in","global","variables","for","all","trainable\/model","variables",".","Example",":","..","code-block",":",":","python","#","Create","the","server","object","like","this",":","hosts","=","[","'host1.com","'",",","'host2.com","'","]","cluster_spec","=","tf.train.ClusterSpec","(","{","'ps","'",":","[","h","+","':2222","'","for","h","in","hosts","]",",","'worker","'",":","[","h","+","':2223","'","for","h","in","hosts","]","}",")","server","=","tf.train.Server","(","cluster_spec",",","job_name=args.job",",","task_index=args.task",",","config=get_default_sess_config","(",")",")","#","initialize","trainer","with","this","server","object","..","code-block",":",":","none","#","Start","training","like","this",":","(","host1",")","$",".\/train.py","--","job","worker","--","task","0","(","host1",")","$","CUDA_VISIBLE_DEVICES=",".\/train.py","--","job","ps","--","task","0","(","host2",")","$",".\/train.py","--","job","worker","--","task","1","(","host2",")","$","CUDA_VISIBLE_DEVICES=",".\/train.py","--","job","ps","--","task","1"]
docstr: Args:
      get_grad_fn (-> [(grad, var)]):
      get_opt_fn (-> tf.train.Optimizer): callable which returns an optimizer
  
  Returns:
      (tf.Operation, tf.Operation, tf.Operation):
  
      1. the training op.
  
      2. the op which sync all the local variables from PS.
      This op should be run before training.
  
      3. the op which sync all the local `MODEL_VARIABLES` from PS.
      You can choose how often to run it by yourself.
docstr_tok: ["Args",":","get_grad_fn","(","-",">","[","(","grad",",","var",")","]",")",":","get_opt_fn","(","-",">","tf.train.Optimizer",")",":","callable","which","returns","an","optimizer","Returns",":","(","tf.Operation",",","tf.Operation",",","tf.Operation",")",":","1.","the","training","op",".","2.","the","op","which","sync","all","the","local","variables","from","PS",".","This","op","should","be","run","before","training",".","3.","the","op","which","sync","all","the","local","`","MODEL_VARIABLES","`","from","PS",".","You","can","choose","how","often","to","run","it","by","yourself","."]
code: def build(self, get_grad_fn, get_opt_fn):
      """
      Args:
          get_grad_fn (-> [(grad, var)]):
          get_opt_fn (-> tf.train.Optimizer): callable which returns an optimizer
  
      Returns:
          (tf.Operation, tf.Operation, tf.Operation):
  
          1. the training op.
  
          2. the op which sync all the local variables from PS.
          This op should be run before training.
  
          3. the op which sync all the local `MODEL_VARIABLES` from PS.
          You can choose how often to run it by yourself.
      """
      with override_to_local_variable():
          get_global_step_var()
  
      get_opt_fn = memoized(get_opt_fn)
      # Build the optimizer first, before entering any tower.
      # This makes sure that learning_rate is a global variable (what we expect)
      get_opt_fn()  # TODO get_opt_fn called before main graph was built
  
      # Ngpu * Nvar * 2
      grad_list = DataParallelBuilder.build_on_towers(
          self.towers,
          get_grad_fn,
          devices=self.raw_devices,
          use_vs=[True] * len(self.towers),
      )  # open vs at each tower
      DataParallelBuilder._check_grad_list(grad_list)
  
      all_grads, all_vars = split_grad_list(grad_list)
      avg_grads = allreduce_grads_naive(
          all_grads, devices=self.raw_devices
      )  # N
      avg_grads = [(g, v) for g, v in zip(all_grads, all_vars[0])]
      with tf.device(self.param_server_device):
          ps_var_grads = (
              DistributedReplicatedBuilder._apply_shadow_vars(avg_grads)
          )
          var_update_ops = self._apply_gradients_and_copy(
              get_opt_fn(), grad_list, ps_var_grads
          )
          self._shadow_vars = [v for (__, v) in ps_var_grads]
          self._shadow_model_vars = (
              DistributedReplicatedBuilder._shadow_model_variables(
                  self._shadow_vars
              )
          )
  
      # TODO add options to synchronize less
      main_fetch = tf.group(*var_update_ops, name="main_fetches")
      train_op = self._add_sync_queues_and_barrier(
          "post_copy_barrier", [main_fetch]
      )
  
      # initial local_vars syncing
      with tf.name_scope("initial_sync_variables"):
          initial_sync_op = self._get_initial_sync_op()
      if len(self._shadow_model_vars) and self.is_chief:
          with tf.name_scope("sync_model_variables"):
              model_sync_op = self._get_sync_model_vars_op()
      else:
          model_sync_op = None
      return train_op, initial_sync_op, model_sync_op
  
imports: import re
  import tensorflow
  from ..tfutils.common import get_global_step_var, get_op_tensor_name
  from ..utils import logger
  from ..utils.argtools import memoized
  from .training import DataParallelBuilder, GraphBuilder
  from .utils import OverrideCachingDevice, split_grad_list, allreduce_grads_naive, override_to_local_variable
label_desc: Use image node_id 4 for calling the DistributedReplicatedBuilder obj's underlying member method code with example usage: obj.build(get_grad_fn, get_opt_fn) and returns: train_op,  initial_sync_op,  model_sync_op