日志¶

FlashInfer 提供了一个日志记录功能，以帮助调试问题和重现崩溃。本文档描述了所有可用的日志记录级别及其功能。

快速入门¶

使用两个环境变量启用日志记录

# Set logging level (0-10)
export FLASHINFER_LOGLEVEL=3

# Set log destination (default is stdout)
export FLASHINFER_LOGDEST=stdout  # or stderr, or a file path like "flashinfer.log"

日志级别¶

级别	名称	功能	用例
0	禁用 (默认)	无日志记录（零开销）	生产
1	函数名称	仅函数名称	基本跟踪
3	输入/输出	函数名称 + 参数 + 带有元数据的输出	标准调试
5	统计信息	级别 3 + 张量统计信息（最小值、最大值、平均值、NaN/Inf 计数）	数值分析
10	飞行记录器 - 完整输入/输出转储	级别 5 + 将所有输入/输出张量转储到 `.pt`（或 `.safetensors`）文件	完全可重现性 / 调试

环境变量¶

主配置¶

变量	类型	默认值	描述
`FLASHINFER_LOGLEVEL`	int	0	日志级别 (0, 1, 3, 5, 10)
`FLASHINFER_LOGDEST`	str	`stdout`	日志目标：`stdout`、`stderr` 或文件路径

转储配置 (级别 10)¶

当 FLASHINFER_LOGLEVEL 设置为 10 时，可以使用以下环境变量来配置转储行为

变量	类型	默认值	描述
`FLASHINFER_DUMP_DIR`	str	`flashinfer_dumps`	保存转储文件的目录
`FLASHINFER_DUMP_MAX_SIZE_GB`	float	20	转储目录的最大大小（以 GB 为单位）
`FLASHINFER_DUMP_MAX_COUNT`	int	1000	转储 API 调用的最大数量
`FLASHINFER_DUMP_INCLUDE`	str	(空)	要包含的逗号分隔模式（fnmatch 样式）
`FLASHINFER_DUMP_EXCLUDE`	str	(空)	逗号分隔的要排除的模式（fnmatch 样式）
`FLASHINFER_DUMP_SAFETENSORS`	int	0	设置为 1 以使用 safetensors 格式（无 pickle，但会丢失步幅信息）

SafeTensors 格式（可选）¶

默认情况下，张量使用 torch.save() 保存，该方法保留张量步幅和连续性信息。为了更快的、无 pickle 的序列化，您可以启用 safetensors 格式

export FLASHINFER_DUMP_SAFETENSORS=1

警告

SafeTensors 不保留张量步幅或非连续性。所有张量都保存为连续的。如果步幅保留对您的调试很重要，请使用默认的 torch.save 格式。

比较:

方面	torch.save (默认)	safetensors
速度	标准	更快
安全性	使用 pickle	不使用 pickle（更安全）
步幅保留	✅ 是	❌ 否（仅连续）
文件扩展名	`.pt`	`.safetensors`
依赖项	torch	需要 `pip install safetensors`

重放与格式无关：重放命令根据文件扩展名自动检测格式。

转储筛选 (包含/排除)¶

使用 FLASHINFER_DUMP_INCLUDE 和 FLASHINFER_DUMP_EXCLUDE 来控制转储哪些 API 调用。这在运行具有许多 API 调用的端到端推理时特别有用，但您只关心特定的调用。

模式语法 (fnmatch 样式)

* 匹配任意数量的字符
? 匹配单个字符
匹配区分大小写
对于类方法，函数名称的格式为 ClassName.method_name

筛选逻辑:

如果设置了 FLASHINFER_DUMP_INCLUDE，则仅转储匹配至少一个模式的 API
如果设置了 FLASHINFER_DUMP_EXCLUDE，则跳过匹配任何模式的 API
两者可以组合：首先应用包含筛选器，然后应用排除筛选器

示例:

# Only dump decode-related APIs
export FLASHINFER_DUMP_INCLUDE="*decode*"

# Dump everything except __init__ and plan methods
export FLASHINFER_DUMP_EXCLUDE="*.__init__,*.plan"

# Only dump run() methods from wrapper classes
export FLASHINFER_DUMP_INCLUDE="*Wrapper.run"

# Dump all single_* APIs except prefill
export FLASHINFER_DUMP_INCLUDE="single_*"
export FLASHINFER_DUMP_EXCLUDE="*prefill*"

# Only dump a specific wrapper's run method
export FLASHINFER_DUMP_INCLUDE="BatchDecodeWithPagedKVCacheWrapper.run"

# Dump FP8 APIs but not quantization steps
export FLASHINFER_DUMP_INCLUDE="*fp8*,*FP8*"
export FLASHINFER_DUMP_EXCLUDE="*quantize*"

常见模式:

模式	匹配
`decode`	`single_decode_with_kv_cache`、`BatchDecodeWithPagedKVCacheWrapper.run`
`*Wrapper.run`	`BatchDecodeWithPagedKVCacheWrapper.run`, `BatchPrefillWithPagedKVCacheWrapper.run`
`*.__init__`	所有 wrapper `__init__` 方法
`*.plan`	所有 wrapper `plan` 方法
`mm_fp8`	`mm_fp8` 函数的精确匹配
`single_*`	`single_decode_with_kv_cache`, `single_prefill_with_kv_cache`

进程 ID 替换¶

在文件路径中使用 %i 进行自动进程 ID 替换（适用于多 GPU 训练）

export FLASHINFER_LOGDEST="flashinfer_log_%i.txt"  # → flashinfer_log_12345.txt

杂项说明和示例¶

CUDA 图兼容性¶

为了避免同步问题，第 5 级统计信息会在 CUDA 图捕获期间被自动跳过。

# This works correctly - no synchronization errors
with torch.cuda.graph(cuda_graph):
    result = mm_fp4(a, b, scales, ...)  # Level 5 logging active
    # Statistics automatically skipped during capture

输出显示：[statistics skipped: CUDA graph capture in progress]

多 GPU 环境中的进程 ID¶

# Use %i for process ID substitution
export FLASHINFER_LOGLEVEL=3
export FLASHINFER_LOGDEST="logs/flashinfer_api_%i.log"

torchrun --nproc_per_node=8 awesome_script_that_uses_FlashInfer.py

# Creates separate logs:
# logs/flashinfer_api_12345.log (rank 0)
# logs/flashinfer_api_12346.log (rank 1)
# ...

第 0 级没有开销¶

在第 0 级，装饰器会返回原始函数，不做任何更改。没有 wrapper，没有检查，没有开销。

飞行记录器和回放¶

FlashInfer 包含一种“飞行记录器”模式（第 10 级），用于捕获输入/输出以实现可重现性。

转储目录结构¶

当启用第 10 级日志记录时，FlashInfer 会创建以下结构

FLASHINFER_DUMP_DIR/
├── session.jsonl                    # Central log: one line per event (quick scanning)
├── 20250108_143216_802_pid12345_mm_fp8_call0001/
│   ├── metadata.jsonl               # Per-dump metadata (JSONL format)
│   ├── inputs.pt                    # Input tensors (or .safetensors if enabled)
│   └── outputs.pt                   # Output tensors (or .safetensors if enabled)
├── 20250108_143216_868_pid12345_single_decode_call0001/
│   ├── metadata.jsonl
│   ├── inputs.pt                    # (or .safetensors)
│   └── outputs.pt                   # (or .safetensors)
└── ...

JSONL 格式：session.jsonl 和 metadata.jsonl 都使用 JSON Lines 格式（每行一个 JSON 对象）。这使得

防崩溃日志记录：每个 API 调用会追加两行（inputs_saved，然后 completed）
快速扫描：使用 session.jsonl 浏览所有记录的调用，无需读取子目录
流式读取：逐行处理记录，用于大型会话

每个转储的 metadata.jsonl:

第 1 行：在执行之前写入（execution_status: "inputs_saved"）
第 2 行：在成功执行之后追加（execution_status: "completed"）

如果发生崩溃，只有第 1 行存在，从而保留输入以进行调试。

中央 session.jsonl:

所有 API 调用的统一日志。使用标准工具进行过滤和分析

# Enable Flight Recorder (Metadata + Tensors)
export FLASHINFER_LOGLEVEL=10
export FLASHINFER_DUMP_DIR=./my_dumps

# Run your application
python3 benchmarks/flashinfer_benchmark.py --routine mm_fp4 --m 4 --n 1024 --k 7168 --out_dtype bfloat16 --backends cudnn --use_128x4_sf_layout --use_nvfp4 --refcheck -vv --generate_repro_command --use_cupti --no_cuda_graph --num_iters 5
... output redacted ...

# Replay recorded calls
export FLASHINFER_LOGLEVEL=0 # 1 for more detailed replay results.
flashinfer replay --dir ./my_dumps
# or
python -m flashinfer replay --dir ./my_dumps

[1] nvfp4_quantize (20251204_143216_802_pid12345_nvfp4_quantize_call0001): ✅ Passed
[2] fp4_quantize (20251204_143216_868_pid12345_fp4_quantize_call0001): ✅ Passed
[3] nvfp4_quantize (20251204_143216_949_pid12345_nvfp4_quantize_call0002): ✅ Passed
[4] fp4_quantize (20251204_143217_003_pid12345_fp4_quantize_call0002): ✅ Passed
[5] mm_fp4 (20251204_143217_178_pid12345_mm_fp4_call0001): ✅ Passed
[6] mm_fp4 (20251204_143217_346_pid12345_mm_fp4_call0002): ✅ Passed
[7] mm_fp4 (20251204_143217_427_pid12345_mm_fp4_call0003): ✅ Passed
[8] mm_fp4 (20251204_143217_475_pid12345_mm_fp4_call0004): ✅ Passed
[9] mm_fp4 (20251204_143217_510_pid12345_mm_fp4_call0005): ✅ Passed
[10] mm_fp4 (20251204_143217_551_pid12345_mm_fp4_call0006): ✅ Passed
[11] mm_fp4 (20251204_143217_591_pid12345_mm_fp4_call0007): ✅ Passed
[12] mm_fp4 (20251204_143217_631_pid12345_mm_fp4_call0008): ✅ Passed
[13] mm_fp4 (20251204_143217_672_pid12345_mm_fp4_call0009): ✅ Passed
[14] mm_fp4 (20251204_143217_708_pid12345_mm_fp4_call0010): ✅ Passed
[15] mm_fp4 (20251204_143217_769_pid12345_mm_fp4_call0011): ✅ Passed
[16] mm_fp4 (20251204_143217_812_pid12345_mm_fp4_call0012): ✅ Passed
[17] mm_fp4 (20251204_143217_852_pid12345_mm_fp4_call0013): ✅ Passed
[18] mm_fp4 (20251204_143217_904_pid12345_mm_fp4_call0014): ✅ Passed
[19] mm_fp4 (20251204_143218_153_pid12345_mm_fp4_call0015): ✅ Passed
[20] mm_fp4 (20251204_143218_390_pid12345_mm_fp4_call0016): ✅ Passed
[21] mm_fp4 (20251204_143218_627_pid12345_mm_fp4_call0017): ✅ Passed
[22] mm_fp4 (20251204_143218_862_pid12345_mm_fp4_call0018): ✅ Passed

Summary: 22 passed, 0 failed/mismatch

基于 Python 的回放示例¶

以下示例演示了如何使用第 10 级日志记录以编程方式使用 Python 转储和回放 API 调用。

示例 1：`bmm_fp8` - 简单函数调用¶

生产者脚本 (bmm_fp8_producer.py)

此脚本初始化张量，调用 bmm_fp8，并将输入/输出转储到磁盘。

"""
Producer script: Run bmm_fp8 with Level 10 logging to dump tensors.

Usage:
    FLASHINFER_LOGLEVEL=10 FLASHINFER_DUMP_DIR=./bmm_fp8_dumps python bmm_fp8_producer.py
"""
import torch
from flashinfer import bmm_fp8

def to_float8(x, dtype=torch.float8_e4m3fn):
    """Convert tensor to FP8 with per-tensor scaling."""
    finfo = torch.finfo(dtype)
    min_val, max_val = x.aminmax()
    amax = torch.maximum(min_val.abs(), max_val.abs()).clamp(min=1e-12)
    scale = finfo.max / amax
    x_scl_sat = (x * scale).clamp(min=finfo.min, max=finfo.max)
    return x_scl_sat.to(dtype), scale.float().reciprocal()

# Parameters
b, m, n, k = 4, 64, 128, 256
input_dtype = torch.float8_e4m3fn
mat2_dtype = torch.float8_e4m3fn
res_dtype = torch.bfloat16

# Create input tensors
input_bf16 = torch.randn([b, m, k], device="cuda", dtype=torch.bfloat16)
input_fp8, input_inv_s = to_float8(input_bf16, dtype=input_dtype)

# mat2: row major -> column major (transposed)
mat2_bf16 = torch.randn([b, n, k], device="cuda", dtype=torch.bfloat16).transpose(-2, -1)
mat2_fp8, mat2_inv_s = to_float8(mat2_bf16, dtype=mat2_dtype)

# Pre-allocate output
res = torch.empty([b, m, n], device="cuda", dtype=res_dtype)

# Call bmm_fp8 - this will be logged/dumped at Level 10
bmm_fp8(input_fp8, mat2_fp8, input_inv_s, mat2_inv_s, res_dtype, res, backend="cublas")

# Print a small portion of the output for verification
print("Output shape:", res.shape)
print("Output[0, :3, :3]:")
print(res[0, :3, :3])

重放脚本 (bmm_fp8_reproducer.py)

此脚本加载转储的张量并重放 bmm_fp8 调用。

"""
Reproducer script: Load dumped tensors and replay bmm_fp8.

Usage:
    python bmm_fp8_reproducer.py
"""
import torch
from pathlib import Path
from flashinfer import bmm_fp8
from flashinfer.api_logging import replay_from_dump

DUMP_DIR = "./bmm_fp8_dumps"

# Find the bmm_fp8 dump directory (should be the only one or the latest)
dump_path = Path(DUMP_DIR)
bmm_dumps = sorted([d for d in dump_path.iterdir() if d.is_dir() and "bmm_fp8" in d.name])
latest_dump = bmm_dumps[-1]  # Use the latest dump
print(f"Loading dump from: {latest_dump}")

# Use replay_from_dump to load inputs and optionally execute
result = replay_from_dump(
    str(latest_dump),
    compare_outputs=True,  # Load expected outputs for comparison
    device="cuda",
    run=False,  # We'll call the function manually below
)

# Extract the loaded arguments - args contains all positional args including the output tensor
args = result["args"]
kwargs = result["kwargs"]
expected_tensors = result.get("expected_tensors", {})

# Replay the call - args already contains (input, mat2, input_inv_s, mat2_inv_s, dtype, out)
res = bmm_fp8(*args, **kwargs)

# Print the same portion for comparison
print("Replayed output shape:", res.shape)
print("Replayed output[0, :3, :3]:")
print(res[0, :3, :3])

# Compare with expected output if available
if "result" in expected_tensors:
    expected = expected_tensors["result"]
    if torch.allclose(res, expected, rtol=1e-3, atol=1e-3):
        print("\n✅ Output matches expected result!")
    else:
        diff = (res - expected).abs().max().item()
        print(f"\n❌ Output mismatch! Max diff: {diff}")

示例 2：`BatchDecodeWithPagedKVCacheWrapper` - 有状态 Wrapper 类¶

生产者脚本 (batch_decode_producer.py)

此脚本演示了使用需要 __init__、plan 和 run 调用的有状态 wrapper 类进行日志记录。

"""
Producer script: Run BatchDecodeWithPagedKVCacheWrapper with Level 10 logging.

Usage:
    FLASHINFER_LOGLEVEL=10 FLASHINFER_DUMP_DIR=./batch_decode_dumps python batch_decode_producer.py
"""
import torch
import flashinfer

# Parameters
batch_size = 4
kv_len = 512
page_size = 16
num_kv_heads = 4
num_qo_heads = 32
head_dim = 128
kv_layout = "NHD"

# Create query tensor
q = torch.randn(batch_size, num_qo_heads, head_dim, device="cuda", dtype=torch.float16)

# Create paged KV cache
num_pages_per_seq = (kv_len + page_size - 1) // page_size
total_num_pages = num_pages_per_seq * batch_size
kv_shape = [total_num_pages, 2, page_size, num_kv_heads, head_dim]  # NHD layout
kv_data = torch.randn(*kv_shape, device="cuda", dtype=torch.float16)

# Create index tensors
kv_indptr = torch.arange(0, batch_size + 1, device="cuda", dtype=torch.int32) * num_pages_per_seq
kv_indices = torch.arange(0, total_num_pages, device="cuda", dtype=torch.int32)
kv_last_page_len = torch.full(
    (batch_size,), (kv_len - 1) % page_size + 1, dtype=torch.int32, device="cuda"
)

# Create workspace and wrapper - __init__ will be logged
workspace_buffer = torch.empty(32 * 1024 * 1024, dtype=torch.int8, device="cuda")
wrapper = flashinfer.decode.BatchDecodeWithPagedKVCacheWrapper(workspace_buffer, kv_layout)

# Plan - will be logged
wrapper.plan(
    kv_indptr,
    kv_indices,
    kv_last_page_len,
    num_qo_heads,
    num_kv_heads,
    head_dim,
    page_size,
    data_type=torch.float16,
    q_data_type=torch.float16,
)

# Run - will be logged
output, lse = wrapper.run(q, kv_data, return_lse=True)

# Print a small portion of the output
print("Output shape:", output.shape)
print("Output[0, :3, :3]:")
print(output[0, :3, :3])
print("\nLSE shape:", lse.shape)
print("LSE[0, :5]:", lse[0, :5])

重放脚本 (batch_decode_reproducer.py)

此脚本演示了重放一系列有状态 API 调用。

"""
Reproducer script: Replay BatchDecodeWithPagedKVCacheWrapper calls.

Usage:
    python batch_decode_reproducer.py
"""
import torch
from pathlib import Path
from flashinfer.api_logging import replay_sequence

DUMP_DIR = "./batch_decode_dumps"

# replay_sequence handles stateful objects automatically via object_registry
# It will:
# 1. Replay __init__ to create the wrapper instance
# 2. Replay plan() on the same instance
# 3. Replay run() on the same instance and compare outputs
results = replay_sequence(DUMP_DIR, device="cuda")

# Print summary
passed = 0
failed = 0
for i, res in enumerate(results):
    func_name = res.get("metadata", {}).get("function_name", "unknown")
    dump_dir = Path(res.get("dump_dir", "")).name

    if "error" in res:
        print(f"[{i+1}] {func_name} ({dump_dir}): ❌ Error: {res['error']}")
        failed += 1
    elif res.get("comparison_match", True):
        print(f"[{i+1}] {func_name} ({dump_dir}): ✅ Passed")
        passed += 1
    else:
        print(f"[{i+1}] {func_name} ({dump_dir}): ❌ Mismatch")
        failed += 1

print(f"\nSummary: {passed} passed, {failed} failed")

# For manual inspection, you can also access individual results
# Find the 'run' call result (usually the last non-init, non-plan call)
for res in results:
    func_name = res.get("metadata", {}).get("function_name", "")
    if "run" in func_name and "execution_result" in res:
        output = res["execution_result"]
        if isinstance(output, tuple):
            output_tensor, lse = output
            print("\nReplayed output[0, :3, :3]:")
            print(output_tensor[0, :3, :3])
            print("Replayed LSE[0, :5]:", lse[0, :5])
        break

不使用 `replay_from_dump` 的手动回放¶

为了获得更多控制，您可以手动加载转储的张量

注意

此示例假定默认 torch.save 格式（.pt 文件）。如果使用 FLASHINFER_DUMP_SAFETENSORS=1 创建了转储，请使用 safetensors.torch.load_file() 代替 torch.load()。

"""
Manual replay: Load tensors directly from .pt files.
"""
import json
import torch
from pathlib import Path
from flashinfer import bmm_fp8

# Path is an example, replace with the actual path.
dump_dir = Path("./bmm_fp8_dumps/20250108_103217_012_pid12345_bmm_fp8_call0001")

# Load metadata from JSONL (read last line for most complete state)
with open(dump_dir / "metadata.jsonl") as f:
    lines = [line.strip() for line in f if line.strip()]
    metadata = json.loads(lines[-1])  # Last line has completed state

print(f"Function: {metadata['function_name']}")
print(f"Module: {metadata['module']}")
print(f"Status: {metadata['execution_status']}")
print(f"Input tensors: {metadata['tensor_info']['input_tensor_keys']}")

# Load input tensors
inputs = torch.load(dump_dir / "inputs.pt", map_location="cuda")

# Load expected outputs (if execution completed successfully)
outputs_path = dump_dir / "outputs.pt"
if outputs_path.exists():
    expected = torch.load(outputs_path, map_location="cuda")
    print(f"Output tensors: {list(expected.keys())}")

# Tensors are ready to use - reconstruct the call as needed
for key, tensor in inputs.items():
    print(f"  {key}: shape={tensor.shape}, dtype={tensor.dtype}")

扫描会话历史记录¶

使用中央 session.jsonl 快速扫描所有记录的 API 调用

"""
Scan session.jsonl for quick overview of recorded calls.
"""
import json
from pathlib import Path
from collections import Counter

dump_root = Path("./my_dumps")
session_file = dump_root / "session.jsonl"

# Read all records
records = []
with open(session_file) as f:
    for line in f:
        if line.strip():
            records.append(json.loads(line))

# Filter to completed calls only
completed = [r for r in records if r["execution_status"] == "completed"]
print(f"Total completed calls: {len(completed)}")

# Count by function name
func_counts = Counter(r["function_name"] for r in completed)
print("\nCalls by function:")
for func, count in func_counts.most_common():
    print(f"  {func}: {count}")

# Find calls that didn't complete (potential crashes)
inputs_only = [r for r in records if r["execution_status"] == "inputs_saved"]
# Group by dump_dir to find incomplete calls
completed_dirs = {r["dump_dir"] for r in completed}
incomplete = [r for r in inputs_only if r["dump_dir"] not in completed_dirs]
if incomplete:
    print(f"\n⚠️  Found {len(incomplete)} incomplete calls (potential crashes):")
    for r in incomplete:
        print(f"  - {r['function_name']} at {r['dump_dir']}")

日志¶

快速入门¶

日志级别¶

环境变量¶

主配置¶

转储配置 (级别 10)¶

SafeTensors 格式（可选）¶

转储筛选 (包含/排除)¶

进程 ID 替换¶

杂项说明和示例¶

CUDA 图兼容性¶

多 GPU 环境中的进程 ID¶

第 0 级没有开销¶

飞行记录器和回放¶

转储目录结构¶

基于 Python 的回放示例¶

示例 1：bmm_fp8 - 简单函数调用¶

示例 2：BatchDecodeWithPagedKVCacheWrapper - 有状态 Wrapper 类¶

不使用 replay_from_dump 的手动回放¶

扫描会话历史记录¶

示例 1：`bmm_fp8` - 简单函数调用¶

示例 2：`BatchDecodeWithPagedKVCacheWrapper` - 有状态 Wrapper 类¶

不使用 `replay_from_dump` 的手动回放¶