code: def push_pull(
      tensor,
      scope="",
      average=None,
      device_dense="",
      device_sparse="",
      compression=Compression.none,
      op=None,
      enable_async=False,
  ):
      """Perform an push_pull on a tf.Tensor or tf.IndexedSlices.
      Arguments:
          tensor: tf.Tensor, tf.Variable, or tf.IndexedSlices to reduce.
                  The shape of the input must be identical across all ranks.
          average:
              .. warning:: .. deprecated
  
                  Use `op` instead. Will be removed.
  
          scope: the graph name scope
          average: If True, computes the average over all ranks.
                   Otherwise, computes the sum over all ranks.
          device_dense: Device to be used for dense tensors. Uses GPU by default.
          device_sparse: Device to be used for sparse tensors. Uses GPU by default.
          compression: Compression algorithm used to reduce the amount of data
                       sent and received by each worker node.  Defaults to not
                       using compression.
          op: The reduction operation to combine tensors across different ranks.
              Defaults to Average if None is given.
  
      Returns:
          A tensor of the same shape and type as `tensor`, summed across all
          processes.
      """
      op = handle_average_backwards_compatibility(op, average).value
      # Averaging happens in framework code, so translate that to Sum for the actual call
      true_op = Sum if op == Average else op
  
      with tf.device(device_dense):
          byteps_size = tf.cast(size(), dtype=tensor.dtype)
          tensor_compressed, ctx = compression.compress(tensor)
          summed_tensor_compressed = _push_pull(
              tensor_compressed, scope
          )
          summed_tensor = compression.decompress(
              summed_tensor_compressed, ctx
          )
          if not enable_async:
              _div = tf.div if hasattr(tf, "div") else tf.math.divide
              new_tensor = (
                  _div(summed_tensor, byteps_size)
                  if op == Average
                  else summed_tensor
              )
          else:  # no need to average for async training
              new_tensor = summed_tensor
      return new_tensor
  
file_path: byteps/byteps/tensorflow/__init__.py
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import os
  import warnings
  from byteps.tensorflow.compression import Compression
  from byteps.tensorflow.ops import broadcast, _push_pull
  from byteps.tensorflow.ops import init, shutdown, suspend, resume, get_pushpull_speed
  from byteps.tensorflow.ops import size, local_size, rank, local_rank
  from byteps.tensorflow.ops import handle_average_backwards_compatibility
  from byteps.tensorflow.util import _executing_eagerly
  import tensorflow
  from tensorflow.python.ops import control_flow_ops
label_desc: Use image node_id 1 for calling a global function with example usage: push_pull(tensor, scope, average, device_dense, device_sparse, compression, op, enable_async) and returns: new_tensor