Apply all-reduce algorithm over specified gradient tensors.

  Args:
    dev_prefixes: list of prefix strings to use to generate PS device names.
    tower_grads: the gradients to reduce.
    num_workers: number of worker processes across entire job.
    alg: the all-reduce algorithm to apply.
    num_shards: alg-specific sharding factor.
    gpu_indices: indices of local GPUs in order usable for ring-reduce.
    agg_small_grads_max_bytes: largest tensor eligible for aggregation,
      in number of bytes.

  Returns:
    list of reduced tensors, packing values

def sum_gradients_all_reduce(dev_prefixes,
                             tower_grads,
                             num_workers,
                             alg,
                             num_shards,
                             gpu_indices,
                             agg_small_grads_max_bytes=0):
    """Apply all-reduce algorithm over specified gradient tensors.

  Args:
    dev_prefixes: list of prefix strings to use to generate PS device names.
    tower_grads: the gradients to reduce.
    num_workers: number of worker processes across entire job.
    alg: the all-reduce algorithm to apply.
    num_shards: alg-specific sharding factor.
    gpu_indices: indices of local GPUs in order usable for ring-reduce.
    agg_small_grads_max_bytes: largest tensor eligible for aggregation,
      in number of bytes.

  Returns:
    list of reduced tensors, packing values
  """
    alg_contains_shuffle = contains_any(alg, ["pscpu", "psgpu"])
    is_hierarchical = "/" in alg
    if "pscpu" in alg:
        aux_devices = [prefix + "/cpu:0" for prefix in dev_prefixes]
    elif "psgpu" in alg:
        aux_devices = [
            prefix + "/gpu:%d" % i for i in range(len(gpu_indices))
            for prefix in dev_prefixes
        ]
    else:
        aux_devices = ["/job:localhost/cpu:0"]
    aux_device_groups = group_device_names(
        aux_devices, num_shards if alg_contains_shuffle else 1)
    group_index = 0
    if agg_small_grads_max_bytes > 0:
        tower_grads, packing = pack_small_tensors(
            tower_grads, max_bytes=agg_small_grads_max_bytes)

    else:
        packing = None
    new_tower_grads = []
    if alg == "better":
        raw_devices = ["/gpu:%i" % (i) for i in gpu_indices]
        agg_grads = aggregate_gradients_using_copy_with_device_selection(
            tower_grads, raw_devices)
        for arr in tower_grads:
            new_tower_grads.append(
                [(g, v) for (_, v), (g, _) in zip(arr, agg_grads)])
    else:
        reduced_gv_list = []
        for grad_and_vars in zip(*tower_grads):
            reduced_gv_list.append(
                sum_grad_and_var_all_reduce(
                    grad_and_vars, num_workers, alg, gpu_indices, aux_devices
                    if is_hierarchical else aux_device_groups[group_index],
                    num_shards))
            group_index = (group_index + 1) % len(aux_device_groups)
        new_tower_grads = [list(x) for x in zip(*reduced_gv_list)]
    return new_tower_grads, packing