################################################################################################# # # Copyright (c) 2023 - 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: BSD-3-Clause # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # # 1. Redistributions of source code must retain the above copyright notice, this # list of conditions and the following disclaimer. # # 2. Redistributions in binary form must reproduce the above copyright notice, # this list of conditions and the following disclaimer in the documentation # and/or other materials provided with the distribution. # # 3. Neither the name of the copyright holder nor the names of its # contributors may be used to endorse or promote products derived from # this software without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE # DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR # SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. # ################################################################################################# """ Tests the high-level Conv2d interface """ from math import ceil import unittest import cutlass import cutlass.utils.datatypes as datatypes from cutlass.backend.utils.device import device_cc from utils import ExpectException import os class Conv2dEquivalence: """ Helper class for testing the equivalence of different constructions of the Conv2d interface """ def __init__(self, conv_kind, element_A, element_B, element_C, element_D, element_accumulator, alignment_A, alignment_B, alignment_C): self.element_A = element_A self.element_B = element_B self.element_C = element_C self.element_D = element_D self.element_accumulator = element_accumulator self.alignment_A = alignment_A self.alignment_B = alignment_B self.alignment_C = alignment_C self.conv_kind = conv_kind self.plan = cutlass.op.Conv2d( kind=self.conv_kind, element_A=element_A, element_B=element_B, element_C=element_C, element_D=element_D, element_accumulator=element_accumulator) self.op = self.plan.construct( alignment_A=self.alignment_A, alignment_B=self.alignment_B, alignment_C=self.alignment_C) def _plans_equal(self, other_plan) -> bool: """ Compares whether two plans are equal :param other_plan: plan to compare against the default Conv2d :type other_plan: cutlass.op.Conv2d :return: whether `other_plan` is equivalent to `self.plan` :rtype: bool """ other_op = other_plan.construct( alignment_A=self.alignment_A, alignment_B=self.alignment_B, alignment_C=self.alignment_C) return self.op.rt_module.emit() == other_op.rt_module.emit() def generic_test(self): """ Tests the equivalence of various constructions of the Conv2d interface when using CUTLASS data types and layouts for constructing the Conv2d interface """ if not datatypes.is_numpy_available(): return # Test when specifying all parameters plan_other = cutlass.op.Conv2d( kind=self.conv_kind, element_A=self.element_A, element_B=self.element_B, element_C=self.element_C, element_D=self.element_D, element_accumulator=self.element_accumulator) assert self._plans_equal(plan_other) # Test when specifying all parameters but A plan_other = cutlass.op.Conv2d( kind=self.conv_kind, element_B=self.element_B, element_C=self.element_C, element_D=self.element_D, element_accumulator=self.element_accumulator, element=self.element_A) assert self._plans_equal(plan_other) # Test when specifying all parameters but A and B as tensors using generic element and output plan_other = cutlass.op.Conv2d( kind=self.conv_kind, element_C=self.element_C, element_D=self.element_D, element_accumulator=self.element_accumulator, element=self.element_A) assert self._plans_equal(plan_other) # Test without explicit accumulator. Only run if the type of C and the accumulator are equal if self.element_C == self.element_accumulator: plan_other = cutlass.op.Conv2d( kind=self.conv_kind, element_C=self.element_C, element_D=self.element_D, element=self.element_A) assert self._plans_equal(plan_other) # Test with only the generic types. Only rune if the types of A, B, C, and D are the same if (self.element_A == self.element_B and self.element_A == self.element_C and self.element_A == self.element_D and self.element_A == self.element_accumulator): plan_other = cutlass.op.Conv2d(kind=self.conv_kind, element=self.element_A) assert self._plans_equal(plan_other) def numpy_test(self): """ Tests the equivalence of various constructions of the Conv2d interface when using numpy as a frontend """ if not datatypes.is_numpy_available(): return import numpy as np type_A = datatypes.numpy_type(self.element_A) type_B = datatypes.numpy_type(self.element_B) type_C = datatypes.numpy_type(self.element_C) type_D = datatypes.numpy_type(self.element_D) type_accum = datatypes.numpy_type(self.element_accumulator) size = (2, 2) A = np.zeros(size, dtype=type_A) B = np.zeros(size, dtype=type_B) C = np.zeros(size, dtype=type_C) D = np.zeros(size, dtype=type_D) return self.tensor_test(type_A, type_B, type_C, type_D, type_accum, A, B, C, D) def torch_test(self): """ Tests the equivalence of various constructions of the Conv2d interface when using torch as a frontend """ if not datatypes.is_torch_available(): return import torch type_A = datatypes.torch_type(self.element_A) type_B = datatypes.torch_type(self.element_B) type_C = datatypes.torch_type(self.element_C) type_D = datatypes.torch_type(self.element_D) type_accum = datatypes.torch_type(self.element_accumulator) size = (2, 2) A = torch.empty(size, dtype=type_A) B = torch.empty(size, dtype=type_B) C = torch.empty(size, dtype=type_C) D = torch.empty(size, dtype=type_D) return self.tensor_test(type_A, type_B, type_C, type_D, type_accum, A, B, C, D) def tensor_test(self, type_A, type_B, type_C, type_D, type_accum, A, B, C, D): # Test when specifying all parameters via tensors plan_np = cutlass.op.Conv2d(kind=self.conv_kind, A=A, B=B, C=C, D=D, element_accumulator=type_accum) assert self._plans_equal(plan_np) # Test when specifying all parameters but A as tensors plan_np = cutlass.op.Conv2d(kind=self.conv_kind, B=B, C=C, D=D, element_accumulator=type_accum, element_A=type_A) assert self._plans_equal(plan_np) # Test when specifying all parameters but A and B as tensors and using generic element and output if type_A == type_B: plan_np = cutlass.op.Conv2d(kind=self.conv_kind, C=C, D=D, element_accumulator=type_accum, element=type_A) assert self._plans_equal(plan_np) # Test without explicit accumulator. Only run if the type of C and the accumulator. if type_C == type_accum: plan_np = cutlass.op.Conv2d(kind=self.conv_kind, A=A, B=B, C=C, D=D) assert self._plans_equal(plan_np) # Test with only the generic types and layouts. Only run if types and layouts of A, B, C, and D are the same. if (type_A == type_B and type_A == type_C and type_A == type_D and type_A == type_accum): plan_np = cutlass.op.Conv2d(kind=self.conv_kind, element=type_A) assert self._plans_equal(plan_np) def test_all(self): """ Runs all tests on the Gemm interface """ self.generic_test() self.numpy_test() self.torch_test() @unittest.skipIf(device_cc() <= 80, 'Device compute capability is insufficient for SM80 tests.') class ConvEquivalenceTest(unittest.TestCase): """ Tests the equivalence of different constructions of the Conv2d interface """ pass type2alignment = { cutlass.DataType.f16: 8, cutlass.DataType.f32: 4 } def add_test(conv_kind, element_A, element_B, element_C, element_D, element_accumulator): test_name = f"test_conv2d_{conv_kind}_{element_A}_{element_B}_{element_C}_{element_D}_{element_accumulator}" def run(self): conv2d_eq = Conv2dEquivalence( conv_kind=conv_kind, element_A=element_A, element_B=element_B, element_C=element_C, element_D=element_D, element_accumulator=element_accumulator, alignment_A=type2alignment[element_A], alignment_B=type2alignment[element_B], alignment_C=type2alignment[element_C] ) conv2d_eq.test_all() setattr(ConvEquivalenceTest, test_name, run) for conv_kind in ["fprop", "wgrad", "dgrad"]: for types in [ [cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f16], [cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f32], [cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f32, cutlass.DataType.f32, cutlass.DataType.f16], [cutlass.DataType.f16, cutlass.DataType.f16, cutlass.DataType.f32, cutlass.DataType.f32, cutlass.DataType.f32], [cutlass.DataType.f32, cutlass.DataType.f32, cutlass.DataType.f32, cutlass.DataType.f32, cutlass.DataType.f32] ]: add_test(conv_kind, types[0], types[1], types[2], types[3], types[4]) @unittest.skipIf(device_cc() <= 80, 'Device compute capability is insufficient for SM80 tests.') class Conv2dErrorTests(unittest.TestCase): """ Tests various error scenarios that arise with the high-level Gemm interface """ def test_alignment(self): """ Tests case in which the alignment specified is unsupported """ plan = cutlass.op.Conv2d(kind="fprop", element=cutlass.DataType.f16) with ExpectException(True, 'Alignment 3 is not supported for F16. The construction should fail.'): op = plan.construct(alignment_A=3, alignment_B=3, alignment_C=3) def test_invalid_tile_description(self): """ Tests scenarios in which an invalid tile description is provided for a given CC """ plan = cutlass.op.Conv2d(kind="fprop", element=cutlass.DataType.f16) td = plan.tile_descriptions()[0] td.threadblock_shape=[17, 32, 5] plan.tile_description = td with ExpectException(True, 'The threadblock shape is invalid. The compilation should fail.'): plan.compile() # Clean up the error message os.remove("./cutlass_python_compilation_device_error.txt") if __name__ == '__main__': unittest.main()