current_sync_states[A][B] = N 的含义：流 A 已经"观察到"了流 B 上序号 ≤ N 的所有操作
关键规则：
- 流创建时（create_stream，第 264-274 行）：
self.host_sync_state[stream] = 0 self.current_sync_states[stream] = self.host_sync_state.copy() - 新流的自身序号初始化为 0，并且只继承 host 当时已知的流状态。已存在的流不会自动获得新流的条目。
默认值为 -1：creation_state.get(usage_stream, -1)。如果字典中没有 usage_stream 这个 key，返回 -1。 - 同步操作传播状态（_state_wait_for_other，第 293-297 行）：
state[stream] = max(state.get(stream, -1), seq_num)
将另一个流的当前 seq_num（至少为 0）写入，取 max 确保只前进不后退。
步骤 1: 创建 default stream (id=0)
host_sync_state = {0: 0}
current_sync_states[0] = {0: 0}
步骤 2: 在 default stream 上分配 tensor x
（tensor 的 allocation_stream = 0）
步骤 3: 在 default stream 上执行 kernel (seq_num=1)
current_sync_states[0] = {0: 1} ← 自增
步骤 4: 创建 stream1
host_sync_state = {0: 0, stream1: 0}
current_sync_states[stream1] = {0: 0, stream1: 0} ← copy from host
此时 current_sync_states[0] 仍是 {0: 1}
注意：stream1 不在 current_sync_states[0] 中！
步骤 5: stream1.wait_stream(default) → stream1 上使用 x
current_sync_states[stream1] = {0: 1, stream1: 0} ← 合并了 stream 0 的状态
（这解决了数据竞争，但 stream 0 仍然不知道 stream1 的存在）
步骤 6: 检查 creation_stream(0) 是否同步了 usage_stream(stream1)
creation_state = current_sync_states[0] = {0: 1}
creation_state.get(stream1, -1) → -1 ← key 不存在，返回默认值
-1 >= 0 → False ← 未同步！需要 record_stream
NPU Sanitizer 功能说明书
1. 概述
NPU Sanitizer 是一个用于检测 NPU 多流程序中同步错误的运行时分析工具。它基于 PyTorch 上游的 CUDA Sanitizer（
torch.cuda._sanitizer）架构，针对昇腾 NPU 进行了适配和扩展。NPU Sanitizer 提供两类独立的检测能力：
2. 启用方式
# 启用基本的数据竞争检测 export TORCH_NPU_SANITIZER=1 # 同时启用 record_stream 检测 export TORCH_NPU_SANITIZER=1 export TORCH_NPU_SANITIZER_CHECK_RECORD_STREAM=1代码方式：
import torch_npu.npu._sanitizer as sanitizer sanitizer.npu_sanitizer.check_record_stream = True sanitizer.npu_sanitizer.enable()3. 整体架构
3.1 关键组件
NPUSanitizer_sanitizer.pyNPUSanitizerDispatchMode_stream_check.pyNPUEventHandler_stream_check.pyEventHandlerCallbackRegistry_npu_trace.pyPyCallbackTriggerPyCallbackTrigger.hStreamAndEventKernelNpu.cpprecord_stream原生函数实现，含 sanitizer 埋点3.2 数据流
4. 功能一：数据竞争检测
4.1 原理
数据竞争发生在不同流上对同一 tensor 进行未同步的并发访问时。Sanitizer 通过维护一个 同步状态矩阵 来判断两个流之间是否已经同步。
同步状态矩阵 (
StreamSynchronizations)核心数据结构：
current_sync_states: dict[StreamId, dict[StreamId, SeqNum]]current_sync_states[A][B] = N表示：流 A 已经"看到"了流 B 上序号 ≤ N 的所有操作。状态更新规则：
current_sync_states[S][S] = seq_num++event.record(stream S)recorded_sync_states[event] = current_sync_states[S].copy()stream A.wait_event(event)current_sync_states[A]合并recorded_sync_states[event]stream_synchronize(S)current_sync_states[S]device_synchronize()数据竞争判定
当流 S 上的 kernel 访问 tensor T 时：
prev_access（包含访问流prev_stream和序号prev_seq）current_sync_states[S].get(prev_stream, -1) >= prev_seq4.2 检测时机与报告
CUDASanitizerErrors异常，程序中断4.3 解决数据竞争的方法
# 方法 1: usage_stream.wait_stream(creation_stream) stream1.wait_stream(default_stream) with torch_npu.npu.stream(stream1): y = x + 1 # 安全：stream1 已同步 default stream # 方法 2: usage_stream.wait_event(event) event = torch_npu.npu.Event() event.record() # 在 default stream 上记录 with torch_npu.npu.stream(stream1): stream1.wait_event(event) y = x + 1 # 安全 # 方法 3: device synchronize（重量级） torch_npu.npu.synchronize()5. 功能二：record_stream 缺失检测
5.1 背景：为什么需要 record_stream
参考 PyTorch 官方文档：
NPU 的 caching allocator 只知道 tensor 在哪个流上分配。当 tensor 在其他流上使用后被释放时，allocator 可能在分配流上立即复用这块内存，但此时使用流上的操作可能还没执行完，造成 use-after-free。
record_stream告诉 allocator："这个 tensor 还在 stream1 上被使用，在 stream1 完成前不要复用它的内存"。5.2 什么时候需要 record_stream
根据 PyTorch 文档，有两种方式保证内存安全：
方式 A：调用
record_stream（让 allocator 管理）x.record_stream(stream1) # allocator 会等 stream1 完成后才复用内存方式 B：手动同步（用户自己管理）
# 确保 creation_stream 等待 usage_stream 完成后，才可能释放/复用内存 default_stream.wait_stream(stream1) # creation waits usage del x # 此时安全，因为 default stream 已经等过 stream1 了关键区分：
usage.wait(creation)creation.wait(usage)device_synchronize()5.3 检测原理
record_stream 检测分两阶段：
阶段一：记录（kernel launch 时）
每次 kernel launch 时，对于访问的每个 tensor，检查当前执行流是否与 tensor 的分配流不同。如果不同，记录一条
CrossStreamUsage，并记下当前全局操作序号seq_num：@dataclass class CrossStreamUsage: usage_stream: int # 使用流 ID seq_num: int # 使用时的全局操作序号 operator: str # 算子名 (如 "aten::add.Tensor") stack_trace: StackSummary # 调用栈如果同一 tensor 在同一流上被多次使用，
seq_num会更新为最后一次使用的序号，确保后续同步检查能覆盖所有使用。阶段二：检查（tensor 释放时或手动 flush）
当 tensor 的内存被释放时（
_handle_memory_deallocation），对每条跨流使用记录检查：其中"creation_stream 已同步等待 usage_stream"的判定复用同步状态矩阵，并比较具体的
seq_num：def _is_creation_stream_synced_to_usage(self, creation_stream, usage_stream, usage_seq_num): creation_state = self.syncs.current_sync_states.get(creation_stream, {}) return creation_state.get(usage_stream, -1) >= usage_seq_num这里比较
>= usage_seq_num而非>= 0，是为了解决两个边界问题：流创建继承的误判：当 creation_stream 晚于 usage_stream 创建时，
create_stream会将host_sync_state（包含{usage_stream: 0}）复制给新流。如果只检查>= 0，这个继承的初始值0会被误判为"已同步"。而实际 kernel 的seq_num从 1 开始（self.seq_num += 1后再调用update_seq_num），所以0 >= 1为 False，正确识别为未同步。同一流多次使用，wait 在中间：如果 tensor 在 usage_stream 上先后执行了 seq=5 和 seq=8 的操作，但
creation.wait_stream(usage)只在 seq=5 之后调用（此时synced_value = 5），则5 >= 8为 False，正确检测到后续使用未被同步覆盖。5.4 为什么延迟到释放时检查
如果在 kernel launch 时就检查，会产生大量误报。因为典型的用户代码是：
with torch_npu.npu.stream(stream1): y = x + 1 # ← 如果在这里检查，sync 还没发生 default_stream.wait_stream(stream1) # ← 用户在后面补上同步 del x # ← 释放时 creation stream 已经 sync 了延迟到释放时检查，此时同步状态矩阵已经反映了所有已发生的同步操作，能准确判断是否需要 record_stream。
5.5 record_stream 追踪路径
record_stream调用通过两条路径被追踪：路径 1：Python Dispatch Mode 拦截
# NPUSanitizerDispatchMode.__torch_dispatch__ if "record_stream" in func_name: # 提取 tensor.data_ptr() 和 stream_id # 调用 event_handler._handle_record_stream(data_ptr, stream_id) return func(*args, **kwargs)路径 2：C++ 原生函数埋点
// StreamAndEventKernelNpu.cpp void NPUNativeFunctions::record_stream(Tensor& self, Stream stream) { // sanitizer trace trigger->traceNpuRecordStream(data_ptr, npu_stream); // 实际的 allocator 注册 NPUCachingAllocator::recordStream(self.storage().data_ptr(), npu_stream); }C++ 层通过
PyCallbackTrigger→NPURecordStreamCallbacks→_handle_record_stream回调到 Python。两条路径都将 stream 添加到
NPUTensorInfo.recorded_streams集合中（set去重，不会冲突）。5.6 检测时机与报告
event_handler.flush_record_stream_warnings()检查所有存活的 tensorevent_handler.record_stream_errors列表5.7 错误输出示例
6. 两类检测的关系
完全正确的多流代码需要同时解决两个问题：
x = torch.randn(100, device='npu') # default stream 创建 stream1 = torch_npu.npu.Stream() # 解决问题 1: 数据竞争 stream1.wait_stream(default_stream) # usage waits creation # 解决问题 2: 内存安全 (二选一) x.record_stream(stream1) # 方式 A: 告诉 allocator # 或: 在后面调用 default_stream.wait_stream(stream1) # 方式 B: 手动管理 with torch_npu.npu.stream(stream1): y = x + 17. 测试矩阵
TestDataRaceDetectionTestMissingRecordStreamTestRecordStreamNotNeededTestRecordStreamNotNeededTestRecordStreamNotNeededTestDataRaceDetectionTestRecordStreamNotNeededTestCorrectUsage8. 核心数据结构
8.1 数据竞争检测（父类
EventHandler）class TensorInfo: allocation_stack_trace: StackSummary | None # 分配时调用栈 reads: list[Access] # 上次写入后的所有读访问 write: Access | None # 最后一次写访问 class Access: type: AccessType # READ 或 WRITE seq_num: int # 操作序号（全局递增） stream: int # 执行流 ID operator: str # 算子 schema stack_trace: StackSummary class StreamSynchronizations: current_sync_states: dict[StreamId, dict[StreamId, SeqNum]] # 流间同步矩阵 recorded_sync_states: dict[EventId, dict[StreamId, SeqNum]] # event 快照 host_sync_state: dict[StreamId, SeqNum] # host 端同步状态8.2 record_stream 检测（
NPUEventHandler扩展）class NPUTensorInfo: allocation_stream: int | None # 分配时的流 ID allocation_stack_trace: StackSummary | None # 分配时调用栈 recorded_streams: set[int] # record_stream 注册过的流 cross_stream_usages: dict[int, CrossStreamUsage] # 跨流使用记录 (key=流ID) class CrossStreamUsage: usage_stream: int # 使用流 ID seq_num: int # 最后一次使用时的全局操作序号 operator: str # 算子名 stack_trace: StackSummary # 使用时调用栈9. seq_num 同步判定详解
9.1 为什么不能用
>= 0StreamSynchronizations在创建新流时，会将host_sync_state复制给新流作为初始状态：def create_stream(self, stream): self.host_sync_state[stream] = 0 self.current_sync_states[stream] = self.host_sync_state.copy()如果 creation_stream 晚于 usage_stream 创建，其初始状态会包含
{usage_stream: 0}。这个0只代表"创建时知道该流存在"，不代表发生过任何同步。而第一个 kernel 的seq_num为 1（self.seq_num += 1后赋值），因此：creation_state[usage_stream]>= 0>= usage_seq_num (≥1)-1（不存在）0（继承）create_stream复制≥ 1（真实同步）wait_stream/device_sync9.2 同一流多次使用的覆盖
_record_cross_stream_usage在遇到已记录的流时，不再跳过，而是更新seq_num：existing = info.cross_stream_usages.get(stream) if existing is not None: existing.seq_num = current_seq # 更新为最新使用的序号 continue这确保检查时比较的是最后一次使用的序号，避免
wait在多次使用中间时的漏检。10. 限制与注意事项
性能开销：Sanitizer 会拦截每个算子调度并收集调用栈，仅建议在调试阶段使用。
record_stream 检测为非阻断式：检测到缺少 record_stream 时只打印警告到 stderr，不中断程序。这是因为检测发生在 C++ 分配器回调中，该回调内部捕获了所有异常。
deallocation 时机依赖 Python GC：tensor 的 Python 对象被销毁后，caching allocator 会标记内存块可复用，但实际的 deallocation 回调触发时机取决于引用计数和垃圾回收。
回调累积：每次
reset_sanitizer+setup_sanitizer会在CallbackRegistry中追加新回调，旧回调仍然存在（但引用已失效的 handler，会静默失败）。在大量测试场景下可能有轻微性能影响。默认流特殊性：默认流 ID 为 0，在
StreamSynchronizations初始化时即创建。NPU 默认流的语义可能与 CUDA 默认流有差异。