# Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved. # Copyright 2020 The HuggingFace Team. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import unittest import numpy as np import paddle import paddle.distributed as dist from paddle.distributed.communication.group import _get_global_group from paddlenlp.transformers.segment_parallel_utils import ( ReshardLayer, split_inputs_sequence_dim, ) def prepare_data(batch_major=True, dim_size=4, batch_size=2, seq_len=2, num_head=2, h=4): assert dim_size == 3 or dim_size == 4, f"dim_size should be 3 or 4, but {dim_size}" batch_size = batch_size seq_len = seq_len h = h num_head = num_head sep = dist.get_world_size() # assert sep == 2, f"sep should be 2, but {sep}" num_elem = batch_size * seq_len // sep * num_head * h // num_head local_rank = dist.get_rank() input_data_list = [] split_tensor_list = [] if dim_size == 4: shape = ( [batch_size, seq_len, num_head // sep, h // num_head] if batch_major else [seq_len, batch_size, num_head // sep, h // num_head] ) split_axis = 1 if batch_major else 0 concat_axis = 2 else: shape = [batch_size, seq_len // sep, h] if batch_major else [seq_len // sep, batch_size, h] split_axis = 2 concat_axis = 1 if batch_major else 0 for rank in range(sep): t = paddle.to_tensor( np.reshape(np.arange(rank * num_elem, (rank + 1) * num_elem) + 1, shape), dtype=paddle.float32 ) input_data_list.append(t) split_tensor_list.append(paddle.split(t, sep, axis=split_axis)) input_data = input_data_list[dist.get_rank()] expected_output_data = [t[local_rank] for t in split_tensor_list] expected_output_data = paddle.concat(expected_output_data, axis=concat_axis) return input_data, expected_output_data def run_forward_backward(x, y_grad, split_axis=0, concat_axis=2): x = x.detach() x.stop_gradient = False reshard_layer = ReshardLayer(sep_group=_get_global_group()) y = reshard_layer( x, split_axis=split_axis, concat_axis=concat_axis, ) paddle.autograd.backward([y], [y_grad], True) return y, x.grad def should_test(sep_degree): if sep_degree <= 1: print(f"sep degree should greater than 1, but is {sep_degree}, skip this test") return sep_degree > 1 class TestReshardLayer(unittest.TestCase): def setUp(self): dist.init_parallel_env() def test_split_inputs(self): batch_size = 8 seq_len = 4096 sep = dist.get_world_size() sep_rank = dist.get_rank() if not should_test(sep): return inputs_ids = paddle.randint(low=0, high=65535, shape=(batch_size, seq_len)) labels = paddle.randint(low=0, high=2, shape=(batch_size, seq_len)) inputs = {"inputs_ids": inputs_ids, "labels": labels} splited_inputs = split_inputs_sequence_dim(inputs, sep_rank, sep) expected_local_inputs = {} for k, v in inputs.items(): expected_local_inputs[k] = paddle.split(v, sep, axis=1)[sep_rank] assert k in splited_inputs np.testing.assert_equal(expected_local_inputs[k].numpy(), splited_inputs[k].numpy()) def test_reshard(self): # [s / sep, b, h] -> [s, b, h / sep] seq_len = 16 sep = dist.get_world_size() if not should_test(sep): return assert seq_len % sep == 0, f"seq_len should be divisible by sep, seq_len:{seq_len}, sep:{sep}" def check_equal(input_data, expected_output_data, split_axis=0, concat_axis=2): bin_bout_output_grad = expected_output_data bin_bout_output, bin_bout_input_grad = run_forward_backward( input_data, bin_bout_output_grad, split_axis=split_axis, concat_axis=concat_axis, ) np.testing.assert_equal(bin_bout_output.numpy(), expected_output_data.numpy()) np.testing.assert_equal(bin_bout_input_grad.numpy(), input_data.numpy()) dim_size = 3 for batch_major in [True, False]: for dim_size in [3, 4]: for batch_size in [1, 2]: for seq_len in [4096, 4096 * 2, 4096 * 4]: for num_head in [32, 64]: for hidden_size in [num_head * 16, num_head * 32]: if dim_size == 3: # check reshard before flash attn, shaped: [b, s, h] input_data, expected_output_data = prepare_data( batch_major=batch_major, dim_size=dim_size, batch_size=batch_size, seq_len=seq_len, num_head=num_head, h=hidden_size, ) split_axis_for_num_head = 2 concat_axis_for_seq = 1 if batch_major else 0 check_equal( input_data, expected_output_data, split_axis=split_axis_for_num_head, concat_axis=concat_axis_for_seq, ) elif dim_size == 4: # check reshard after flash attn, shaped: [b, s, num_head, head_dim] input_data, expected_output_data = prepare_data( batch_major=batch_major, dim_size=dim_size, batch_size=batch_size, seq_len=seq_len, num_head=num_head, h=hidden_size, ) split_axis_for_seq = 1 if batch_major else 0 concat_axis_for_num_head = 2 check_equal( input_data, expected_output_data, split_axis=split_axis_for_seq, concat_axis=concat_axis_for_num_head, ) if __name__ == "__main__": unittest.main()