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: Append collective ops after the gradient is calculated.
docstr_tok: ["Append","collective","ops","after","the","gradient","is","calculated","."]
code: def _collect_dense_gradients(self, graph_item, var_op_name):
      """Append collective ops after the gradient is calculated."""
      if self.num_replicas * self.num_workers <= 1:
          raise ValueError(
              "CollectiveOps requires collective group size > 1"
          )
  
      compressors = defaultdict(
          lambda: Compressor.create(self._compressor_type, var_op_name)
      )
  
      conf = CollectiveOpsConfig()
      conf.group_size = len(self.all_canonical_replica_devices)
      conf.group_key = get_collective_keys().get_group_key(
          self.all_canonical_replica_devices
      )
      conf.instance_key = get_collective_keys().get_instance_key(
          var_op_name
      )
      conf.merge_op = "Add"
      conf.final_op = "Div"
      if self._spec:
          setattr(conf, "communication_hint", self._spec)
  
      for i in range(0, self.num_replicas):
          op_name = ops.prepend_name_scope(
              var_op_name, replica_prefix(i)
          )
          graph_item.updated = True
          grad, _, _ = graph_item.var_op_name_to_grad_info_v2[op_name]
          # TODO (Tairui): (3) Merge of reduction for performance
          grad_consumers = get_consumers(
              grad.op
          )  # this line must happen before the reduction
  
          # "\/" is added for name scope reuse
          with ops.name_scope(
              replica_prefix(i)
              + "/collective-group-{}/".format(self._group)
          ):
              with ops.colocate_with(grad.op):
                  reduced_grad = compressors[i].reduce(grad, conf)
          update_consumers(grad_consumers, grad, reduced_grad)
  
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 3 for calling the AllReduceSynchronizer obj's underlying member method code with example usage: obj._collect_dense_gradients(graph_item, var_op_name) without return types