class_docstr: AllReduce Synchronizer.
  
  This AllReduce Synchronizer currently uses TensorFlow's `collective_device_ops`
  to insert their AllReduce ops into our graph.
  
  The class AllReduceSynchronizer class contains the following possible instantiations:
  
  1. spec=`auto`: single-node multiple devices, or cross-node AllReduce based on collective ops
  2. spec=`nccl`: single-node multiple devices, or cross-node AllReduce based on NCCL
  3. spec=`ring`/'tree', AllReduce with different reduction structures: ring, tree, etc.
  
  However note that it does not contain the following instantiations:
  
  1. shuffle reduce (reduce to CPU or GPU as in PS) + AllReduce across nodes
  2. any other types of hybrid reduction of PS and AllReduce.
class_docstr_tok: ["AllReduce","Synchronizer",".","This","AllReduce","Synchronizer","currently","uses","TensorFlow","'s","`","collective_device_ops","`","to","insert","their","AllReduce","ops","into","our","graph",".","The","class","AllReduceSynchronizer","class","contains","the","following","possible","instantiations",":","1.","spec=","`","auto","`",":","single-node","multiple","devices",",","or","cross-node","AllReduce","based","on","collective","ops","2.","spec=","`","nccl","`",":","single-node","multiple","devices",",","or","cross-node","AllReduce","based","on","NCCL","3.","spec=","`","ring","`","\/'tree","'",",","AllReduce","with","different","reduction","structures",":","ring",",","tree",",","etc",".","However","note","that","it","does","not","contain","the","following","instantiations",":","1.","shuffle","reduce","(","reduce","to","CPU","or","GPU","as","in","PS",")","+","AllReduce","across","nodes","2.","any","other","types","of","hybrid","reduction","of","PS","and","AllReduce","."]
docstr: Perform in-graph synchronization based on AllReduce and TensorFlow Collective Ops.
  
  Note that collective ops now only supports dense tensors.
  
  Args:
      graph_item (graph_item.GraphItem): the graph_item to be distributed
      var_name (str): the corresponded variable name
  
  Returns:
      graph_item.GraphItem: The new graph
docstr_tok: ["Perform","in-graph","synchronization","based","on","AllReduce","and","TensorFlow","Collective","Ops",".","Note","that","collective","ops","now","only","supports","dense","tensors",".","Args",":","graph_item","(","graph_item.GraphItem",")",":","the","graph_item","to","be","distributed","var_name","(","str",")",":","the","corresponded","variable","name","Returns",":","graph_item.GraphItem",":","The","new","graph"]
code: def in_graph_apply(self, graph_item, var_name):
      """
      Perform in-graph synchronization based on AllReduce and TensorFlow Collective Ops.
  
      Note that collective ops now only supports dense tensors.
  
      Args:
          graph_item (graph_item.GraphItem): the graph_item to be distributed
          var_name (str): the corresponded variable name
  
      Returns:
          graph_item.GraphItem: The new graph
      """
      # Skip allreduce synchronizer when rank <= 1
      if self.num_replicas * self.num_workers <= 1:
          return graph_item
  
      item = graph_item
      var_op_name = get_op_name(var_name)
  
      # Throw an error if the variable is sparse
      master_op_name = ops.prepend_name_scope(
          var_op_name, replica_prefix(0)
      )
      graph_item.updated = True
      grad, _, _ = graph_item.var_op_name_to_grad_info_v2[
          master_op_name
      ]
      graph_item.var_queried.append(master_op_name)
      with item.graph.as_default():
          self._share_initializer(item, var_op_name, master_replica=0)
          if isinstance(grad, ops.IndexedSlices):
              self._collect_sparse_gradients(item, var_op_name)
          else:
              self._collect_dense_gradients(item, var_op_name)
      return item
  
imports: from collections import defaultdict
  from tensorflow.python import ops
  from tensorflow.python.framework import device_spec
  from tensorflow.python.ops import collective_ops
  import autodist
  from autodist.const import ENV
  from autodist.kernel.common.utils import get_consumers, update_consumers, replica_prefix, get_control_consumers, update_control_consumers
  from autodist.kernel.common.utils import get_op_name
  from autodist.kernel.synchronization.collective_key import get_collective_keys
  from autodist.kernel.synchronization.compressor import Compressor, CollectiveOpsConfig
  from autodist.kernel.synchronization.synchronizer import Synchronizer
  from autodist.proto import synchronizers_pb2
  from autodist.utils import logging
label_desc: Use image node_id 2 for calling the AllReduceSynchronizer obj's underlying member method code with example usage: obj.in_graph_apply(graph_item, var_name) and returns: item, graph_item