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: Adds additional ops needed for synchronous distributed training into current graph.
  
  Main purpose of additional ops are:
  1. Initialization
  2. Synchronization
  3. Gradient aggregation
  
  Args:
      graph_item (graph_item.GraphItem): the graph
      var_update_op: The op
      variable_replicator: The dictionary of master variable op name
          -> list of replicated variables, could be None
  
  Returns:
      None
docstr_tok: ["Adds","additional","ops","needed","for","synchronous","distributed","training","into","current","graph",".","Main","purpose","of","additional","ops","are",":","1",".","Initialization","2",".","Synchronization","3",".","Gradient","aggregation","Args",":","graph_item","(","graph_item.GraphItem",")",":","the","graph","var_update_op",":","The","op","variable_replicator",":","The","dictionary","of","master","variable","op","name","-",">","list","of","replicated","variables",",","could","be","None","Returns",":","None"]
code: def add_sync_op(
      self, graph_item, var_update_op, variable_replicator=None
  ):
      """
      Adds additional ops needed for synchronous distributed training into current graph.
  
      Main purpose of additional ops are:
      1. Initialization
      2. Synchronization
      3. Gradient aggregation
  
      Args:
          graph_item (graph_item.GraphItem): the graph
          var_update_op: The op
          variable_replicator: The dictionary of master variable op name
              -> list of replicated variables, could be None
  
      Returns:
          None
      """
      this_worker_cpu = device_spec.DeviceSpecV2.from_string(
          self.worker_device
      )
      this_worker_cpu = this_worker_cpu.replace(
          device_type="CPU", device_index=0
      )
  
      var_op = var_update_op.inputs[UPDATE_OP_VAR_POS].op
      is_trainable = var_op in graph_item.trainable_var_op_to_var
      source_op = self._get_optimizer_source_op(var_update_op)
      cc = get_control_consumers(source_op)
  
      with ops.device(var_op.device):
          if self._staleness == 0:
              queue_ops = self._get_queue_ops(
                  var_update_op, source_op, self.is_chief, is_trainable
              )
          elif self._staleness > 0:
              queue_ops = self._get_queue_ops_stale(
                  var_update_op, source_op, self.is_chief, is_trainable
              )
          else:
              raise ValueError(
                  "staleness should be greater than or equal to 0."
              )
  
          # Only dense trainable variables are replicated locally
          if variable_replicator:
              mirror_variable_update_ops = (
                  variable_replicator.get_all_update_ops(
                      queue_ops, worker_device=this_worker_cpu
                  )
              )
              with ops.device(this_worker_cpu):
                  finish_op = control_flow_ops.group(
                      *mirror_variable_update_ops
                  )
          else:
              finish_op = control_flow_ops.group(*queue_ops)
  
      # Place computation ops of aggregated gradients on PS
      # Note that even though this is doing a graph traversal, it is called in such a way that it
      # only traverses from a gradient aggregator op to a gradient application op (or vice versa) --
      # these corresponding ops should always be adjacent in the graph.
      self._place_post_grad_agg_ops(
          device_spec.DeviceSpecV2.from_string(self.target_device),
          self._var_op_to_agg_grad,
          {var_op: var_update_op} if is_trainable else {},
      )
  
      # Replace the control input of train_op to be finish_op
      # Note(Hao): this cc is stale, i.e. cc \subset get_control_consumers(source_op)
      update_control_consumers(cc, source_op, finish_op)
  
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 8 for calling the PSSynchronizer obj's underlying member method code with example usage: obj.add_sync_op(graph_item, var_update_op, variable_replicator) without return types