Reduce the gradients, apply them with the optimizer,
        and set self.grads to #GPU number of lists of (g, v), containing the all-reduced gradients on each device.

        Args:
            grad_list ([[(grad, var), ...], ...]): #GPU lists to be reduced. Each is the gradients computed on each GPU.
            get_opt_fn (-> tf.train.Optimizer): callable which returns an optimizer

        Returns:
            (tf.Operation, tf.Operation)

            1. the training op.

            2. the op which sync variables from GPU 0 to other GPUs.
                It has to be run before the training has started.
                And you can optionally run it later to sync non-trainable variables.

def build(self, grad_list, get_opt_fn):
        """
        Reduce the gradients, apply them with the optimizer,
        and set self.grads to #GPU number of lists of (g, v), containing the all-reduced gradients on each device.

        Args:
            grad_list ([[(grad, var), ...], ...]): #GPU lists to be reduced. Each is the gradients computed on each GPU.
            get_opt_fn (-> tf.train.Optimizer): callable which returns an optimizer

        Returns:
            (tf.Operation, tf.Operation)

            1. the training op.

            2. the op which sync variables from GPU 0 to other GPUs.
                It has to be run before the training has started.
                And you can optionally run it later to sync non-trainable variables.
        """
        assert len(grad_list) == len(self.towers)
        raw_devices = ['/gpu:{}'.format(k) for k in self.towers]

        DataParallelBuilder._check_grad_list(grad_list)

        dtypes = set([x[0].dtype.base_dtype for x in grad_list[0]])
        dtypes_nccl_supported = [tf.float32, tf.float64]
        if get_tf_version_tuple() >= (1, 8):
            dtypes_nccl_supported.append(tf.float16)
        valid_for_nccl = all([k in dtypes_nccl_supported for k in dtypes])
        if self._mode == 'nccl' and not valid_for_nccl:
            logger.warn("Cannot use mode='nccl' because some gradients have unsupported types. Fallback to mode='cpu'")
            self._mode = 'cpu'

        if self._mode in ['nccl', 'hierarchical']:
            all_grads, all_vars = split_grad_list(grad_list)
            # use allreduce from tf-benchmarks
            # from .batch_allreduce import AllReduceSpecAlgorithm
            # algo = AllReduceSpecAlgorithm('nccl', list(range(8)), 0, 10)
            # all_grads, warmup_ops = algo.batch_all_reduce(all_grads, 1, True, False)
            # print("WARMUP OPS", warmup_ops)

            if self._mode == 'nccl':
                all_grads = allreduce_grads(all_grads, average=self._average)  # #gpu x #param
            else:
                packer = GradientPacker(len(raw_devices))
                succ = packer.compute_strategy(all_grads[0])
                if succ:
                    packed_grads = packer.pack_all(all_grads, raw_devices)
                    packed_grads_aggr = allreduce_grads_hierarchical(
                        packed_grads, raw_devices, average=self._average)
                    all_grads = packer.unpack_all(packed_grads_aggr, raw_devices)
                else:
                    all_grads = allreduce_grads_hierarchical(all_grads, raw_devices, average=self._average)

            self.grads = merge_grad_list(all_grads, all_vars)
        elif self._mode == 'cpu':
            agg_grad_and_vars = aggregate_grads(
                grad_list, colocation=False,
                devices=['/cpu:0'], average=self._average)    # #param x 2
            self.grads = []  # #gpu x #param x 2
            for grad_and_vars in grad_list:   # grad_and_vars: #paramx2
                # take v from each tower, and g from average.
                self.grads.append(
                    [(g, v) for (_, v), (g, _) in zip(grad_and_vars, agg_grad_and_vars)])

        train_ops = []
        opt = get_opt_fn()
        with tf.name_scope('apply_gradients'):
            for idx, grad_and_vars in enumerate(self.grads):
                with tf.device(raw_devices[idx]):
                    # apply_gradients may create variables. Make them LOCAL_VARIABLES
                    with override_to_local_variable(enable=idx > 0):
                        train_ops.append(opt.apply_gradients(
                            grad_and_vars, name='apply_grad_{}'.format(idx)))
        train_op = tf.group(*train_ops, name='train_op')

        with tf.name_scope('sync_variables'):
            post_init_op = SyncMultiGPUReplicatedBuilder.get_post_init_ops()
        return train_op, post_init_op