| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | import pytest |
| 4 | import torch |
| 5 | |
| 6 | from vllm.platforms import CpuArchEnum, current_platform |
| 7 | |
| 8 | if not current_platform.is_cpu(): |
| 9 | pytest.skip("skipping CPU-only tests", allow_module_level=True) |
| 10 | |
| 11 | from vllm._custom_ops import cpu_gemm_wna16 # noqa: E402 |
| 12 | from vllm.model_executor.kernels.linear.mixed_precision.cpu import ( # noqa: E402 |
| 13 | _get_isa_hint, |
| 14 | ) |
| 15 | |
| 16 | # Isolated runs of cpu_gemm_wna16 can hang unless AMX tiles are initialized. |
| 17 | # Guard so non-AMX hosts still take the vec/rvv path. |
| 18 | if torch.cpu._is_amx_tile_supported(): |
| 19 | torch.cpu._init_amx() |
| 20 | |
| 21 | |
| 22 | @pytest.mark.cpu_test |
| 23 | def test_cpu_gemm_wna16_3d_input(): |
| 24 | """cpu_gemm_wna16 flattens 3D [B, T, K] to 2D and reshapes the output back.""" |
| 25 | if not hasattr(torch.ops._C, "cpu_gemm_wna16"): |
| 26 | pytest.skip("cpu_gemm_wna16 op is not available") |
| 27 | |
| 28 | B, T, K, N = 2, 5, 64, 64 |
| 29 | pack_factor = 8 |
| 30 | group_size = 32 |
| 31 | dtype = torch.bfloat16 |
| 32 | q_weight = torch.zeros(N // 16, K * 16 // pack_factor, dtype=torch.int32) |
| 33 | scales = torch.ones(K // group_size, N, dtype=dtype) |
| 34 | |
| 35 | x_3d = torch.randn(B, T, K, dtype=dtype) |
| 36 | x_2d = x_3d.reshape(-1, K) |
| 37 | isa_hint = _get_isa_hint(dtype) |
| 38 | kwargs = dict( |
| 39 | q_weight=q_weight, |
| 40 | scales=scales, |
| 41 | zeros=None, |
| 42 | bias=None, |
| 43 | pack_factor=pack_factor, |
| 44 | isa_hint=isa_hint, |
| 45 | ) |
| 46 | |
| 47 | out_3d = cpu_gemm_wna16(input=x_3d, **kwargs) |
| 48 | out_2d = cpu_gemm_wna16(input=x_2d, **kwargs) |
| 49 | |
| 50 | assert out_3d.shape == (B, T, N) |
| 51 | assert out_2d.shape == (B * T, N) |
| 52 | torch.testing.assert_close(out_3d.reshape(-1, N), out_2d) |
| 53 | |
| 54 | |
| 55 | MODELS = [ |
| 56 | "TheBloke/TinyLlama-1.1B-Chat-v1.0-AWQ", |
| 57 | "Qwen/Qwen1.5-0.5B-Chat-GPTQ-Int4", # without g_idx |
| 58 | "RedHatAI/Qwen3-1.7B-quantized.w4a16", # with zp |
| 59 | "OPEA/Qwen2.5-0.5B-Instruct-int4-sym-inc", |
| 60 | "Qwen/Qwen3-0.6B-FP8", # FP8 W8A16 block-quantized linear |
| 61 | "Qwen/Qwen3-30B-A3B-FP8", # FP8 W8A16 block-quantized MoE |
| 62 | "openai/gpt-oss-20b", # MXFP4 W4A16 |
| 63 | "QuixiAI/Qwen3-30B-A3B-AWQ", # AWQ W4A16 MoE |
| 64 | "Qwen/Qwen3-30B-A3B-GPTQ-Int4", # GPTQ W4A16 MoE |
| 65 | "RedHatAI/Qwen3-30B-A3B-quantized.w4a16", # compressed-tensors W4A16 MoE |
| 66 | ] |
| 67 | DTYPE = ["bfloat16"] |
| 68 | |
| 69 | |
| 70 | @pytest.mark.skipif( |
| 71 | current_platform.get_cpu_architecture() != CpuArchEnum.POWERPC, |
| 72 | reason="PowerPC-only test", |
| 73 | ) |
| 74 | def test_cpu_wna16_power_bf16_uses_vsx(): |
| 75 | """Use the POWER10 BF16 MMA microkernel instead of generic vectors.""" |
| 76 | assert _get_isa_hint(torch.bfloat16) == "vsx" |
| 77 | assert _get_isa_hint(torch.float16) == "vec" |
| 78 | |
| 79 | |
| 80 | @pytest.mark.parametrize("model", MODELS) |
| 81 | @pytest.mark.parametrize("dtype", DTYPE) |
| 82 | def test_cpu_quant(vllm_runner, model, dtype): |
| 83 | with vllm_runner(model, dtype=dtype) as llm: |
| 84 | output = llm.generate_greedy(["The capital of France is"], max_tokens=32) |
| 85 | assert output |
| 86 | print(output) |