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: Get queue operations for staleness synchronous parameter update.
  
  Maintain a list of queues of size equal to <staleness>. At the beginning of each call of this function
  (either by the chief worker or other workers), it checks whether each queue is not full. If yes, it pushes
  a token to each queue. If not, it does nothing (a no_op).
  Then, for the current worker that calls this function, it dequeues a token from its corresponding queue
  (indexed by its worker id).
  The potential enqueue operations and definite dequeue operation are grouped together, and have to be
  finished before the model moves on to the next step.
  As at each invocation of this function, a row of empty space in the list of queues will be filled. Thus
  <staleness> number of consecutive dequeue operations can be done by a worker without blocking, achieving
  stale synchronous parameter update with maximum <staleness> steps difference.
  
  Args:
      var_update_op: The op
  
  Returns:
      A list of queue operations.
docstr_tok: ["Get","queue","operations","for","staleness","synchronous","parameter","update",".","Maintain","a","list","of","queues","of","size","equal","to","<","staleness",">",".","At","the","beginning","of","each","call","of","this","function","(","either","by","the","chief","worker","or","other","workers",")",",","it","checks","whether","each","queue","is","not","full",".","If","yes",",","it","pushes","a","token","to","each","queue",".","If","not",",","it","does","nothing","(","a","no_op",")",".","Then",",","for","the","current","worker","that","calls","this","function",",","it","dequeues","a","token","from","its","corresponding","queue","(","indexed","by","its","worker","id",")",".","The","potential","enqueue","operations","and","definite","dequeue","operation","are","grouped","together",",","and","have","to","be","finished","before","the","model","moves","on","to","the","next","step",".","As","at","each","invocation","of","this","function",",","a","row","of","empty","space","in","the","list","of","queues","will","be","filled",".","Thus","<","staleness",">","number","of","consecutive","dequeue","operations","can","be","done","by","a","worker","without","blocking",",","achieving","stale","synchronous","parameter","update","with","maximum","<","staleness",">","steps","difference",".","Args",":","var_update_op",":","The","op","Returns",":","A","list","of","queue","operations","."]
code: def _get_queue_ops_stale(
      self,
      var_update_op: ops.Operation,
      source_op: ops.Operation,
      is_chief: bool,
      is_trainable: bool,
  ) -> List[ops.Operation]:
      """
      Get queue operations for staleness synchronous parameter update.
  
      Maintain a list of queues of size equal to <staleness>. At the beginning of each call of this function
      (either by the chief worker or other workers), it checks whether each queue is not full. If yes, it pushes
      a token to each queue. If not, it does nothing (a no_op).
      Then, for the current worker that calls this function, it dequeues a token from its corresponding queue
      (indexed by its worker id).
      The potential enqueue operations and definite dequeue operation are grouped together, and have to be
      finished before the model moves on to the next step.
      As at each invocation of this function, a row of empty space in the list of queues will be filled. Thus
      <staleness> number of consecutive dequeue operations can be done by a worker without blocking, achieving
      stale synchronous parameter update with maximum <staleness> steps difference.
  
      Args:
          var_update_op: The op
  
      Returns:
          A list of queue operations.
      """
      var_op = var_update_op.inputs[UPDATE_OP_VAR_POS].op
  
      var_update_sync_queues = [
          data_flow_ops.FIFOQueue(
              self._staleness,
              [dtypes.bool],
              shapes=None,
              name="%s_update_sync_queue_%d" % (var_op.name, i),
              shared_name="%s_update_sync_queue_%d" % (var_op.name, i),
          )
          for i in range(self.num_workers)
      ]
  
      # Enqueue one token to every queue if all queues are not full.
      def _enqueue_row_op():
          enqueue_ops = []
          for q in var_update_sync_queues:
              enqueue_ops.append(q.enqueue(False))
          enqueue_a_row_ops = control_flow_ops.group(*enqueue_ops)
          return enqueue_a_row_ops
  
      def _no_op():
          return gen_control_flow_ops.no_op()
  
      switch_cond = gen_array_ops.identity(True)
      for q in var_update_sync_queues:
          switch_cond = gen_math_ops.logical_and(
              switch_cond,
              gen_math_ops.less(
                  q.size(), gen_array_ops.identity(self._staleness)
              ),
          )
  
      enqueue_a_row_ops = control_flow_ops.cond(
          switch_cond, _enqueue_row_op, _no_op
      )
  
      queue_ops = [enqueue_a_row_ops]
  
      if is_chief:
          if is_trainable:
              var_update_deps = [
                  self._var_op_to_accum_apply_op[var_op],
                  source_op,
              ]
          else:
              var_update_deps = [var_update_op]
          with ops.control_dependencies(var_update_deps):
              dequeue = var_update_sync_queues[self.worker_id].dequeue()
      else:
          # wait for execution of var_update_op
          if is_trainable:
              with ops.control_dependencies(
                  [self._var_op_to_accum_apply_op[var_op]]
              ):
                  dequeue = var_update_sync_queues[
                      self.worker_id
                  ].dequeue()
          else:
              dequeue = var_update_sync_queues[self.worker_id].dequeue()
      queue_ops.append(dequeue)
  
      return queue_ops
  
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 10 for calling the PSSynchronizer obj's underlying member method code with example usage: obj._get_queue_ops_stale(var_update_op, source_op, is_chief, is_trainable) and returns: queue_ops, enqueue_a_row_ops, gen_control_flow_ops