code: def my_collective_reduce(
      input,
      group_size,
      group_key,
      instance_key,
      merge_op,
      final_op,
      subdiv_offsets,
      wait_for=[],
      communication_hint="auto",
      name=None,
  ):
      r"""Mutually reduces multiple tensors of identical type and shape.
  
      Args:
        input: A `Tensor`. Must be one of the following types: `float32`, `half`, `float64`, `int32`, `int64`.
        group_size: An `int`.
        group_key: An `int`.
        instance_key: An `int`.
        merge_op: A `string` from: `"Min", "Max", "Mul", "Add"`.
        final_op: A `string` from: `"Id", "Div"`.
        subdiv_offsets: A list of `ints`.
        wait_for: An optional list of `ints`. Defaults to `[]`.
        communication_hint: An optional `string`. Defaults to `"auto"`.
        name: A name for the operation (optional).
  
      Returns:
        A `Tensor`. Has the same type as `input`.
      """
      _ctx = _context._context or _context.context()
      tld = _ctx._thread_local_data
      if tld.is_eager:
          try:
              _result = _pywrap_tensorflow.TFE_Py_FastPathExecute(
                  _ctx._context_handle,
                  tld.device_name,
                  "CollectiveReduce",
                  name,
                  tld.op_callbacks,
                  input,
                  "group_size",
                  group_size,
                  "group_key",
                  group_key,
                  "instance_key",
                  instance_key,
                  "merge_op",
                  merge_op,
                  "final_op",
                  final_op,
                  "subdiv_offsets",
                  subdiv_offsets,
                  "wait_for",
                  wait_for,
                  "communication_hint",
                  communication_hint,
              )
              return _result
          except _core._FallbackException:
              try:
                  return collective_reduce_eager_fallback(
                      input,
                      group_size=group_size,
                      group_key=group_key,
                      instance_key=instance_key,
                      merge_op=merge_op,
                      final_op=final_op,
                      subdiv_offsets=subdiv_offsets,
                      wait_for=wait_for,
                      communication_hint=communication_hint,
                      name=name,
                      ctx=_ctx,
                  )
              except _core._SymbolicException:
                  pass  # Add nodes to the TensorFlow graph.
          except _core._NotOkStatusException as e:
              _ops.raise_from_not_ok_status(e, name)
      # Add nodes to the TensorFlow graph.
      group_size = _execute.make_int(group_size, "group_size")
      group_key = _execute.make_int(group_key, "group_key")
      instance_key = _execute.make_int(instance_key, "instance_key")
      merge_op = _execute.make_str(merge_op, "merge_op")
      final_op = _execute.make_str(final_op, "final_op")
      if not isinstance(subdiv_offsets, (list, tuple)):
          raise TypeError(
              "Expected list for 'subdiv_offsets' argument to "
              "'collective_reduce' Op, not %r." % subdiv_offsets
          )
      subdiv_offsets = [
          _execute.make_int(_i, "subdiv_offsets")
          for _i in subdiv_offsets
      ]
      if wait_for is None:
          wait_for = []
      if not isinstance(wait_for, (list, tuple)):
          raise TypeError(
              "Expected list for 'wait_for' argument to "
              "'collective_reduce' Op, not %r." % wait_for
          )
      wait_for = [_execute.make_int(_i, "wait_for") for _i in wait_for]
      if communication_hint is None:
          communication_hint = "auto"
      communication_hint = _execute.make_str(
          communication_hint, "communication_hint"
      )
  
      _, _, _op, _outputs = my_new_wrapper_around_byteps_push_pull(
          input=input,
          group_size=group_size,
          group_key=group_key,
          instance_key=instance_key,
          merge_op=merge_op,
          final_op=final_op,
          subdiv_offsets=subdiv_offsets,
          wait_for=wait_for,
          communication_hint=communication_hint,
          name=name,
      )
      _result = _outputs[:]
      if _execute.must_record_gradient():
          _attrs = (
              "T",
              _op._get_attr_type("T"),
              "group_size",
              _op._get_attr_int("group_size"),
              "group_key",
              _op._get_attr_int("group_key"),
              "instance_key",
              _op._get_attr_int("instance_key"),
              "merge_op",
              _op.get_attr("merge_op"),
              "final_op",
              _op.get_attr("final_op"),
              "subdiv_offsets",
              _op.get_attr("subdiv_offsets"),
              "wait_for",
              _op.get_attr("wait_for"),
              "communication_hint",
              _op.get_attr("communication_hint"),
          )
          _inputs_flat = _op.inputs
          _execute.record_gradient(
              "CollectiveReduce", _inputs_flat, _attrs, _result
          )
      (_result,) = _result
      return _result
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import collections
  import enum
  import six
  from tensorflow.python.client import device_lib
  from tensorflow.python.distribute import cross_device_utils
  from tensorflow.python.distribute import device_util
  from tensorflow.python.distribute import reduce_util
  from tensorflow.python.distribute import values
  from tensorflow.python.eager import context
  from tensorflow.python.framework import kernels
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor_util
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import resource_variable_ops
  from tensorflow.python.platform import tf_logging
  from tensorflow.python.util.tf_export import tf_export
  from tensorflow.tools.docs import doc_controls
  import tensorflow
  from tensorflow.python.distribute.cross_device_ops import CollectiveAllReduce
  import tensorflow.python.distribute.cross_device_ops
  from byteps.tensorflow.ops import broadcast, _push_pull
  from byteps.tensorflow import local_rank, size
label_desc: Use image node_id 9 for calling a global function with example usage: my_collective_reduce(input, group_size, group_key, instance_key, merge_op, final_op, subdiv_offsets, wait_for, communication_hint, name) and returns: _result, _result, collective_reduce_eager_fallback