def get_global_norm_of_tensors(input_tensors, norm_type=2, mpu=None, use_graph=False):
    """Get norm of an iterable of tensors.

    This is adapted from torch.nn.utils.clip_grad.clip_grad_norm_ and
    added functionality to handle model parallel parameters. Taken from Nvidia Megatron.

    Arguments:
        input_tensors (Iterable[Tensor]): an iterable of Tensors will have norm computed
        norm_type (float or int): type of the used p-norm. Can be ``'inf'`` for
            infinity norm.

    Returns:
        Total norm of the tensors (viewed as a single vector).
    """
    assert isinstance(input_tensors, Iterable), f'expected Iterable type not {type(input_tensors)}'
    assert all([torch.is_tensor(t) for t in input_tensors]), f'expected list of only tensors'

    norm_type = float(norm_type)
    if norm_type == inf:
        total_norm = max(t.data.abs().max() for t in input_tensors)
        total_norm_cuda = get_accelerator().FloatTensor([float(total_norm)])
        if mpu is not None:
            dist.all_reduce(total_norm_cuda, op=dist.ReduceOp.MAX, group=mpu.get_model_parallel_group())
            total_norm = total_norm_cuda[0].item()
    else:
        if use_graph:
            if 'norm_tensors_compute_buffer' not in graph_cache:
                graph_cache['norm_tensors_compute_buffer'] = [t.data.float().norm(norm_type) for t in input_tensors]
            compute_buffer = graph_cache['norm_tensors_compute_buffer']

            def _norm_tensors(tensor_list, _compute_buffer, _norm_type):
                for i, t in enumerate(tensor_list):
                    _compute_buffer[i].data.copy_(t.data.float().norm(_norm_type)**_norm_type)
                    if i != 0:
                        _compute_buffer[0].data.add_(_compute_buffer[i].data)

            graph_process(False, _norm_tensors, input_tensors, compute_buffer, norm_type)

            total_norm = compute_buffer[0]
        else:
            total_norm = sum([t.data.float().norm(norm_type).item()**norm_type for t in input_tensors])

        total_norm_cuda = get_accelerator().FloatTensor([float(total_norm)]).detach()
        if mpu is not None:
            dist.all_reduce(total_norm_cuda, op=dist.ReduceOp.SUM, group=mpu.get_model_parallel_group())
        total_norm = total_norm_cuda[0].item()**(1. / norm_type)

    if total_norm == float('inf') or total_norm == -float('inf') or total_norm != total_norm:
        total_norm = -1

    return total_norm
Get norm of an iterable of tensors.

This is adapted from torch.nn.utils.clip_grad.clip_grad_norm_ and
added functionality to handle model parallel parameters. Taken from Nvidia Megatron.

Arguments:
input_tensors (Iterable[Tensor]): an iterable of Tensors will have norm computed
norm_type (float or int): type of the used p-norm. Can be ``'inf'`` for
infinity norm.

Returns:
Total norm of the tensors (viewed as a single vector).
opensora/adaptor/utils.py