class_docstr: PS Synchronizer.
  
  Synchronizes gradient updates using a Parameter Server.
  
  For in-graph synchronization, this just aggregates gradients on a worker's CPU.
  
  For between-graph synchronization, this aggregates gradients on a pre-defined
  (defined in the Strategy) parameter server.
  
  To keep this gradient aggregation in sync, the chief gives each worker a token
  for each variable for the workers to mark when their variable update is complete.
docstr: Apply in-graph synchronization to the grad and target in the graph.
  
  Args:
      graph_item (graph_item.GraphItem): The graph to put the new ops in.
      old_update_op (Op): The update op corresponding to the grad and target.
      old_grad: The gradient object.
      old_target: The target tensor.
  
  Returns:
      graph_item.GraphItem
docstr_tok: ["Apply","in-graph","synchronization","to","the","grad","and","target","in","the","graph",".","Args",":","graph_item","(","graph_item.GraphItem",")",":","The","graph","to","put","the","new","ops","in",".","old_update_op","(","Op",")",":","The","update","op","corresponding","to","the","grad","and","target",".","old_grad",":","The","gradient","object",".","old_target",":","The","target","tensor",".","Returns",":","graph_item.GraphItem"]
code: def _aggregate_gradients(
      self, graph_item, old_update_op, old_grad, old_target
  ):
      """
      Apply in-graph synchronization to the grad and target in the graph.
  
      Args:
          graph_item (graph_item.GraphItem): The graph to put the new ops in.
          old_update_op (Op): The update op corresponding to the grad and target.
          old_grad: The gradient object.
          old_target: The target tensor.
  
      Returns:
          graph_item.GraphItem
      """
  
      def ctrl_consumers(op):
          return op._control_outputs  # pylint: disable=protected-access
  
      # Hierarchical reduction at local node
      reduce_to_device = device_spec.DeviceSpecV2.from_string(
          self.worker_device
      ).replace(device_type="CPU", device_index=0)
  
      # TODO(Hao): rethink this -- is this correct?
      graph_item.graph.get_operation_by_name(
          old_target.op.name
      )._set_device_from_string(self.target_device)
      graph_item.graph.get_operation_by_name(
          old_update_op.name
      )._set_device_from_string(self.target_device)
  
      # Aggregate old_grad
      if isinstance(old_grad, ops.Tensor):
          # Dense gradient
          consumer_ops = graph_item.get_ops_in_graph(
              old_grad.consumers()
          )
          ctrl_consumer_ops = graph_item.get_ops_in_graph(
              ctrl_consumers(old_grad.op)
          )
          agg_grad = self._get_aggregated_dense_grad(
              graph_item, old_grad.name, reduce_to_device
          )
  
          # Make gradients consumers to consume the aggregated gradients.
          self._update_gradient_consumers(
              graph_item,
              consumer_ops,
              ctrl_consumer_ops,
              old_grad.name,
              agg_grad,
          )
      elif isinstance(old_grad, ops.IndexedSlices):
          # Sparse gradient
          indices_c_ops = graph_item.get_ops_in_graph(
              old_grad.indices.consumers()
          )
          indices_cc_ops = graph_item.get_ops_in_graph(
              ctrl_consumers(old_grad.indices.op)
          )
          values_c_ops = graph_item.get_ops_in_graph(
              old_grad.values.consumers()
          )
          values_cc_ops = graph_item.get_ops_in_graph(
              ctrl_consumers(old_grad.values.op)
          )
          agg_grad = self._get_aggregated_sparse_grad(
              graph_item, old_target.op, old_grad, reduce_to_device
          )
  
          self._update_gradient_consumers(
              graph_item,
              indices_c_ops,
              indices_cc_ops,
              old_grad.indices.name,
              agg_grad.indices,
          )
          self._update_gradient_consumers(
              graph_item,
              values_c_ops,
              list(set(values_cc_ops).difference(indices_cc_ops)),
              old_grad.values.name,
              agg_grad.values,
          )
      else:
          raise RuntimeError("Incorrect old_grad.")
      return agg_grad
  
imports: from functools import partial
  from typing import List
  from tensorflow.python import ops
  from tensorflow.python.framework import device_spec, dtypes, constant_op
  from tensorflow.python.platform import tf_logging
  from tensorflow.python.ops import math_ops, data_flow_ops, gen_control_flow_ops, control_flow_ops, gen_math_ops, gen_array_ops
  from autodist.const import MAX_INT64, AUTODIST_PREFIX
  from autodist.kernel.common import utils, variable_utils
  from autodist.kernel.common.op_info import UPDATE_OP_VAR_POS
  from autodist.kernel.common.proxy_variable import ProxyVariable
  from autodist.kernel.common.utils import get_op_name, get_consumers, get_ancestors, traverse, update_consumers, update_control_consumers, replica_prefix, strip_replica_prefix, get_control_consumers, remove_from_control_consumers, get_index_from_tensor_name, update_colocation_group
  from autodist.kernel.common.variable_utils import get_read_var_ops
  from autodist.kernel.synchronization.synchronizer import Synchronizer
  from autodist.proto import synchronizers_pb2
label_desc: Use image node_id 4 for calling the PSSynchronizer obj's underlying member method code with example usage: obj._aggregate_gradients(graph_item, old_update_op, old_grad, old_target) and returns: agg_grad, op