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 ps synchronization.
  
  Args:
      graph_item: the old graph item
      var_name: the variable name w/o replica prefix
  
  Returns:
      graph_item.GraphItem
docstr_tok: ["Apply","in-graph","ps","synchronization",".","Args",":","graph_item",":","the","old","graph","item","var_name",":","the","variable","name","w\/o","replica","prefix","Returns",":","graph_item.GraphItem"]
code: def in_graph_apply(self, graph_item, var_name):
      """
      Apply in-graph ps synchronization.
  
      Args:
          graph_item: the old graph item
          var_name: the variable name w/o replica prefix
  
      Returns:
          graph_item.GraphItem
  
      """
      item = graph_item
      var_op_name = get_op_name(var_name)
      master_replica_index = 0
  
      with item.graph.as_default():
          self._prune_control_dependencies(
              item, var_op_name, master_replica=master_replica_index
          )
          self._share_variable(
              item, var_op_name, master_replica=master_replica_index
          )
          master_var_name = ops.prepend_name_scope(
              var_name, replica_prefix(master_replica_index)
          )
          master_var_op_name = get_op_name(master_var_name)
          item.updated = True
          grad, target, update_op = item.var_op_name_to_grad_info_v2[
              master_var_op_name
          ]
          item.var_queried.append(master_var_op_name)
          agg_grad = self._aggregate_gradients(
              item,
              old_update_op=update_op,
              old_grad=grad,
              old_target=target,
          )
  
      # update grad_target_pair and variable info
      for i in range(self.num_replicas):
          var_name_to_remove = ops.prepend_name_scope(
              var_name, replica_prefix(i)
          )
          item.pop_gradient_info(var_name=var_name_to_remove)
          if i != master_replica_index:
              item.info.pop_variable(var_name=var_name_to_remove)
      item.extend_gradient_info(
          grads=[agg_grad],
          targets=[item.graph.get_tensor_by_name(master_var_name)],
      )
      # TODO(Hao): Prune the graph to use unnecessary nodes
      return item
  
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 2 for calling the PSSynchronizer obj's underlying member method code with example usage: obj.in_graph_apply(graph_item, var_name) and returns: item