# Copyright (c) 2022 PaddlePaddle Authors. 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 from op_test import ( OpTest, convert_float_to_uint16, get_device_place, is_custom_device, ) import paddle from paddle.base import core def Heaviside_grad(x, y, dout, astype="float16", is_bfloat16=False): tmp = np.zeros(x.shape).astype(astype) dx = np.multiply(tmp, dout) dy = np.multiply(np.equal(x, 0), dout).astype(astype) if is_bfloat16: dx = convert_float_to_uint16(dx) dy = convert_float_to_uint16(dy) return dx, dy class TestElementwiseOp(OpTest): def setUp(self): self.op_type = "elementwise_heaviside" x = np.random.random((13, 17)).astype("float64") y = np.random.random((13, 17)).astype("float64") self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = {'X': x, 'Y': y} self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} def test_check_output(self): self.check_output( check_pir=True, check_prim_pir=True, check_symbol_infer=False ) def test_check_grad_normal(self): self.check_grad(['X', 'Y'], 'Out', check_pir=True, check_prim_pir=True) def test_check_grad_ignore_x(self): self.check_grad( ['Y'], 'Out', no_grad_set=set("X"), check_pir=True, check_prim_pir=True, ) def test_check_grad_ignore_y(self): self.check_grad( ['X'], 'Out', no_grad_set=set('Y'), check_pir=True, check_prim_pir=True, ) class TestHeavisideBroadcast(unittest.TestCase): def setUp(self): self.input_1 = np.random.rand(2, 100, 13, 17).astype("float32") self.input_2 = np.random.rand(100, 13, 17).astype("float32") self.input_3 = np.random.rand(100, 13, 1).astype("float32") self.input_4 = np.random.rand(13, 17).astype("float32") self.input_5 = np.random.rand(1).astype("float32") self.np_expected1 = np.heaviside(self.input_1, self.input_2) self.np_expected2 = np.heaviside(self.input_2, self.input_3) self.np_expected3 = np.heaviside(self.input_2, self.input_4) self.np_expected4 = np.heaviside(self.input_4, self.input_5) def test_broadcast(self): paddle.disable_static() self.tensor_1 = paddle.to_tensor(self.input_1) self.tensor_2 = paddle.to_tensor(self.input_2) self.tensor_3 = paddle.to_tensor(self.input_3) self.tensor_4 = paddle.to_tensor(self.input_4) self.tensor_5 = paddle.to_tensor(self.input_5) res = paddle.heaviside(self.tensor_1, self.tensor_2) res = res.numpy() np.testing.assert_allclose(res, self.np_expected1, rtol=1e-05) res = paddle.heaviside(self.tensor_2, self.tensor_3) res = res.numpy() np.testing.assert_allclose(res, self.np_expected2, rtol=1e-05) res = paddle.heaviside(self.tensor_2, self.tensor_4) res = res.numpy() np.testing.assert_allclose(res, self.np_expected3, rtol=1e-05) res = paddle.heaviside(self.tensor_4, self.tensor_5) res = res.numpy() np.testing.assert_allclose(res, self.np_expected4, rtol=1e-05) class TestHeavisideAPI_float64(unittest.TestCase): def setUp(self): self.x_np = np.random.random((13, 17)).astype("float64") self.y_np = np.random.random((13, 17)).astype("float64") self.out_np = np.heaviside(self.x_np, self.y_np) self.dtype = "float64" def test_static(self): for use_cuda in ( [False, True] if (paddle.device.is_compiled_with_cuda() or is_custom_device()) else [False] ): place = get_device_place() if use_cuda else paddle.CPUPlace() paddle.enable_static() prog = paddle.static.Program() with paddle.static.program_guard(prog): x = paddle.static.data( name=f"x_{self.dtype}", shape=[13, 17], dtype=self.dtype ) y = paddle.static.data( name=f"y_{self.dtype}", shape=[13, 17], dtype=self.dtype ) out = paddle.heaviside(x, y) exe = paddle.static.Executor(place=place) (res,) = exe.run( prog, feed={ f"x_{self.dtype}": self.x_np, f"y_{self.dtype}": self.y_np, }, fetch_list=out, use_prune=True, ) np.testing.assert_allclose(res, self.out_np, rtol=1e-05) def test_dygraph(self): for use_cuda in ( [False, True] if (paddle.device.is_compiled_with_cuda() or is_custom_device()) else [False] ): place = get_device_place() if use_cuda else paddle.CPUPlace() paddle.disable_static(place=place) result = paddle.heaviside( paddle.to_tensor(self.x_np), paddle.to_tensor(self.y_np) ) np.testing.assert_allclose(result.numpy(), self.out_np, rtol=1e-05) class TestHeavisideAPI_float32(TestHeavisideAPI_float64): def setUp(self): self.x_np = np.random.random((13, 17)).astype("float32") self.y_np = np.random.random((13, 17)).astype("float32") self.out_np = np.heaviside(self.x_np, self.y_np) self.dtype = "float32" class TestHeavisideAPI_int64(TestHeavisideAPI_float64): def setUp(self): self.x_np = np.random.random((13, 17)).astype("int64") self.y_np = np.random.random((13, 17)).astype("int64") self.out_np = np.heaviside(self.x_np, self.y_np) self.dtype = "int64" class TestHeavisideAPI_int32(TestHeavisideAPI_float64): def setUp(self): self.x_np = np.random.random((13, 17)).astype("int32") self.y_np = np.random.random((13, 17)).astype("int32") self.out_np = np.heaviside(self.x_np, self.y_np) self.dtype = "int32" class TestElementwiseOp1(TestElementwiseOp): def setUp(self): self.op_type = "elementwise_heaviside" x = np.random.random(100).astype("float64") y = np.random.random((13, 100)).astype("float64") self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = {'X': x, 'Y': y} self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} class TestElementwiseOp2(TestElementwiseOp): def setUp(self): self.op_type = "elementwise_heaviside" x = np.random.random((13, 100)).astype("float64") y = np.random.random(100).astype("float64") self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = {'X': x, 'Y': y} self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} class TestElementwiseOp3(TestElementwiseOp): def setUp(self): self.op_type = "elementwise_heaviside" x = np.random.uniform(-10, 10, [100]).astype("float64") y = np.random.uniform(-10, 10, [3, 100]).astype("float64") self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = {'X': x, 'Y': y} self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} class TestElementwiseOp4(TestElementwiseOp): def setUp(self): self.op_type = "elementwise_heaviside" x = np.random.uniform(0, 10, []).astype("float64") y = np.random.uniform(-10, 0, [2, 3, 20]).astype("float64") self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = {'X': x, 'Y': y} self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} class TestHeavisideFP16Op(OpTest): def setUp(self): self.dtype = np.float16 self.op_type = "elementwise_heaviside" self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = { 'X': np.random.uniform(1, 2, [20, 5]).astype("float16"), 'Y': np.random.uniform(1, 2, [20, 5]).astype("float16"), } self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} def test_check_output(self): self.check_output( check_pir=True, check_prim_pir=True, check_symbol_infer=False ) def test_check_grad(self): self.check_grad( ['X', 'Y'], 'Out', user_defined_grads=Heaviside_grad( self.inputs['X'], self.inputs['Y'], 1 / self.inputs['X'].size ), check_pir=True, check_prim_pir=True, ) @unittest.skipIf( not (core.is_compiled_with_cuda() or is_custom_device()) or not core.is_bfloat16_supported(get_device_place()), "core is not compiled with CUDA or not support bfloat16", ) class TestHeavisideBF16Op(OpTest): def setUp(self): self.dtype = np.uint16 self.np_dtype = np.float32 self.op_type = "elementwise_heaviside" self.python_api = paddle.heaviside self.prim_op_type = "comp" self.public_python_api = paddle.heaviside self.inputs = { 'X': np.random.uniform(1, 2, [20, 5]).astype(self.np_dtype), 'Y': np.random.uniform(1, 2, [20, 5]).astype(self.np_dtype), } self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])} self.place = get_device_place() self.inputs['X'] = convert_float_to_uint16(self.inputs['X']) self.inputs['Y'] = convert_float_to_uint16(self.inputs['Y']) self.outputs['Out'] = convert_float_to_uint16(self.outputs['Out']) def test_check_output(self): self.check_output_with_place( self.place, check_pir=True, check_prim_pir=True, check_symbol_infer=False, ) def test_check_grad(self): self.check_grad_with_place( self.place, ['X', 'Y'], 'Out', user_defined_grads=Heaviside_grad( self.inputs['X'], self.inputs['Y'], 1 / self.inputs['X'].size, self.np_dtype, True, ), check_pir=True, check_prim_pir=True, ) class TestHeavisideError(unittest.TestCase): def test_input(self): paddle.disable_static() def test_input_x(): paddle.heaviside(1, paddle.randn([100])) self.assertRaises(ValueError, test_input_x) def test_input_y(): paddle.heaviside(paddle.randn([100]), 1) self.assertRaises(ValueError, test_input_y) def test_input_xy(): paddle.heaviside( paddle.randn([100], 'float32'), paddle.randn([100], 'float64') ) self.assertRaises(ValueError, test_input_xy) @unittest.skipIf( not (core.is_compiled_with_cuda() or is_custom_device()), "core is not compiled with CUDA", ) class TestElementwiseHeavisideOp_Stride(OpTest): no_need_check_grad = True def setUp(self): self.op_type = "elementwise_heaviside" self.python_api = paddle.heaviside self.public_python_api = paddle.heaviside self.transpose_api = paddle.transpose self.as_stride_api = paddle.as_strided self.init_dtype() self.init_input_output() self.inputs_stride = { 'X': OpTest.np_dtype_to_base_dtype(self.x), 'Y': OpTest.np_dtype_to_base_dtype(self.y_trans), } self.inputs = { 'X': OpTest.np_dtype_to_base_dtype(self.x), 'Y': OpTest.np_dtype_to_base_dtype(self.y), } self.outputs = {'Out': self.out} def init_dtype(self): self.dtype = np.float64 self.val_dtype = np.float64 def test_check_output(self): place = get_device_place() self.check_strided_forward = True self.check_output( place, ) def init_input_output(self): self.strided_input_type = "transpose" self.x = np.random.uniform(0.1, 1, [13, 17]).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [13, 17]).astype(self.dtype) self.out = np.heaviside(self.x, self.y) self.perm = [1, 0] self.y_trans = np.transpose(self.y, self.perm) def test_check_gradient(self): pass class TestElementwiseHeavisideOp_Stride1(TestElementwiseHeavisideOp_Stride): def init_input_output(self): self.strided_input_type = "transpose" self.x = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype) self.out = np.heaviside(self.x, self.y) self.perm = [0, 1, 3, 2] self.y_trans = np.transpose(self.y, self.perm) class TestElementwiseHeavisideOp_Stride2(TestElementwiseHeavisideOp_Stride): def init_input_output(self): self.strided_input_type = "transpose" self.x = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype) self.out = np.heaviside(self.x, self.y) self.perm = [0, 2, 1, 3] self.y_trans = np.transpose(self.y, self.perm) class TestElementwiseHeavisideOp_Stride3(TestElementwiseHeavisideOp_Stride): def init_input_output(self): self.strided_input_type = "transpose" self.x = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [20, 2, 13, 1]).astype(self.dtype) self.out = np.heaviside(self.x, self.y) self.perm = [0, 1, 3, 2] self.y_trans = np.transpose(self.y, self.perm) class TestElementwiseHeavisideOp_Stride4(TestElementwiseHeavisideOp_Stride): def init_input_output(self): self.strided_input_type = "transpose" self.x = np.random.uniform(0.1, 1, [1, 2, 13, 17]).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [20, 2, 13, 1]).astype(self.dtype) self.out = np.heaviside(self.x, self.y) self.perm = [1, 0, 2, 3] self.y_trans = np.transpose(self.y, self.perm) class TestElementwiseHeavisideOp_Stride5(TestElementwiseHeavisideOp_Stride): def init_input_output(self): self.strided_input_type = "as_stride" self.x = np.random.uniform(0.1, 1, [23, 10, 1, 17]).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [23, 2, 13, 20]).astype(self.dtype) self.y_trans = self.y self.y = self.y[:, 0:1, :, 0:1] self.out = np.heaviside(self.x, self.y) self.shape_param = [23, 1, 13, 1] self.stride_param = [520, 260, 20, 1] class TestElementwiseHeavisideOp_Stride_ZeroDim1( TestElementwiseHeavisideOp_Stride ): def init_input_output(self): self.strided_input_type = "transpose" self.x = np.random.uniform(0.1, 1, []).astype(self.dtype) self.y = np.random.uniform(0.1, 1, [13, 17]).astype(self.dtype) self.out = np.heaviside(self.x, self.y) self.perm = [1, 0] self.y_trans = np.transpose(self.y, self.perm) class TestElementwiseHeavisideOp_Stride_ZeroSize1( TestElementwiseHeavisideOp_Stride ): def init_data(self): self.strided_input_type = "transpose" self.x = np.random.rand(1, 0, 2).astype('float32') self.y = np.random.rand(3, 0, 1).astype('float32') self.out = np.heaviside(self.x, self.y) self.perm = [2, 1, 0] self.y_trans = np.transpose(self.y, self.perm) @unittest.skipIf( not (core.is_compiled_with_cuda() or is_custom_device()), "core is not compiled with CUDA", ) class TestHeavisideZeroSizeTensor(unittest.TestCase): """Regression test for 0-size tensor in paddle.heaviside forward and backward. When input has a dimension of 0, the broadcast backward kernels (ElemwiseGradBroadcast1CUDA / ElemwiseGradBroadcast2CUDA) were incorrectly launched with block_size=0 or grid_size=0, causing CUDA error(9) (cudaErrorInvalidConfiguration). Fix: add early-return guards in ElemwiseGradBroadcast1CUDA (h==0 || w==0) and ElemwiseGradBroadcast2CUDA (pre==0 || n==0 || post==0). """ def setUp(self): self.place = get_device_place() paddle.disable_static(place=self.place) def _check_forward_backward(self, x_shape, y_shape, dtype='float32'): """Run forward + backward and assert output shape and no CUDA error.""" x = paddle.zeros(x_shape, dtype=dtype) y = paddle.ones(y_shape, dtype=dtype) x.stop_gradient = False y.stop_gradient = False out = paddle.heaviside(x, y) expected_shape = list( np.broadcast_shapes(tuple(x_shape), tuple(y_shape)) ) self.assertEqual(list(out.shape), expected_shape) out_grad = paddle.ones_like(out) grads = paddle.grad( [out], [x, y], grad_outputs=[out_grad], allow_unused=True, ) self.assertEqual(list(grads[0].shape), x_shape) self.assertEqual(list(grads[1].shape), y_shape) # Verify no sticky CUDA error was left by any kernel launch core.eager._for_test_check_cuda_error() return out def _check_forward_only(self, x_shape, y_shape, dtype='int32'): """Run forward-only for non-float dtypes and assert shape + no CUDA error.""" x = paddle.zeros(x_shape, dtype=dtype) y = paddle.ones(y_shape, dtype=dtype) out = paddle.heaviside(x, y) expected_shape = list( np.broadcast_shapes(tuple(x_shape), tuple(y_shape)) ) self.assertEqual(list(out.shape), expected_shape) core.eager._for_test_check_cuda_error() return out # --------------------------------------------------------------- # Same-shape 0-size (no broadcast) — ElemwiseGradComputeNoBroadcast # --------------------------------------------------------------- def test_same_shape_zero_leading_dim_float32(self): """[0, 2048] x [0, 2048] – same shape, no broadcast.""" self._check_forward_backward([0, 2048], [0, 2048]) def test_same_shape_zero_leading_dim_float64(self): """[0, 17] x [0, 17].""" self._check_forward_backward([0, 17], [0, 17], 'float64') def test_same_shape_zero_trailing_dim_float64(self): """[13, 0] x [13, 0].""" self._check_forward_backward([13, 0], [13, 0], 'float64') def test_same_shape_zero_trailing_dim_int32(self): """[13, 0] x [13, 0] – int32, forward only.""" self._check_forward_only([13, 0], [13, 0], 'int32') def test_same_shape_zero_trailing_dim_int64(self): """[13, 0] x [13, 0] – int64, forward only.""" self._check_forward_only([13, 0], [13, 0], 'int64') def test_same_shape_zero_leading_dim_int32(self): """[0, 17] x [0, 17] – int32, forward only.""" self._check_forward_only([0, 17], [0, 17], 'int32') def test_same_shape_zero_leading_dim_int64(self): """[0, 17] x [0, 17] – int64, forward only.""" self._check_forward_only([0, 17], [0, 17], 'int64') # --------------------------------------------------------------- # ElemwiseGradBroadcast1CUDA — h=0 (block_size would be 0) # --------------------------------------------------------------- def test_broadcast1_zero_trailing_dim_scalar(self): """[300, 0] x [1] → Broadcast1CUDA(h=pre=0, w=n=1), block_size=0.""" self._check_forward_backward([300, 0], [1]) def test_broadcast1_zero_leading_dim_scalar(self): """[0, 2048] x [1] → Broadcast1CUDA(h=pre=0, w=n=1), block_size=0.""" self._check_forward_backward([0, 2048], [1]) def test_broadcast1_zero_leading_dim_last_dim(self): """[0, 2048] x [2048] → Broadcast1CUDA(h=pre=0, w=n=2048), block_size=0.""" self._check_forward_backward([0, 2048], [2048]) def test_broadcast1_scalar_zero_trailing_dim(self): """[1] x [300, 0] – symmetric of test_broadcast1_zero_trailing_dim_scalar.""" self._check_forward_backward([1], [300, 0]) def test_broadcast1_scalar_zero_leading_dim(self): """[1] x [0, 2048] – symmetric of test_broadcast1_zero_leading_dim_scalar.""" self._check_forward_backward([1], [0, 2048]) def test_broadcast1_last_dim_zero_leading_dim(self): """[2048] x [0, 2048] – symmetric of test_broadcast1_zero_leading_dim_last_dim.""" self._check_forward_backward([2048], [0, 2048]) # --------------------------------------------------------------- # ElemwiseGradBroadcast1CUDA — w=0 (grid_size would be 0) # --------------------------------------------------------------- def test_broadcast1_zero_mid_dim_w_zero(self): """[2, 0, 3] x [0, 3] → Broadcast1CUDA(h=2, w=0), grid_size=0.""" self._check_forward_backward([2, 0, 3], [0, 3]) # --------------------------------------------------------------- # ElemwiseGradBroadcast2CUDA — post=0 (block_size would be 0) # --------------------------------------------------------------- def test_broadcast2_zero_post_dim(self): """[2, 3, 0] x [3, 1] → Broadcast2CUDA(pre=2, n=3, post=0), block_size=0.""" self._check_forward_backward([2, 3, 0], [3, 1]) if __name__ == '__main__': unittest.main()