1.概述

1.1简介

当前taskqueue的设置为一个常量（一次性从环境变量读取后不允许更改），由于taskqueue=2的独立内存池设计暂时无法支持ACLgraph，本提案支持在Python侧灵活控制taskqueue设置的能力，具体为：
(1) 新增Python接口用户设置和获取taskqueue状态值（一个原子变量，默认-1），设置时需要清queue；
(2) 在get_task_enable的公共接口中，先读取该全局变量的值，如果为非-1，则以该值的结果为准，不再从环境变量读取；
(3) 环境变量的读取仍然是一次性的返回常量，保证默认状态下的性能
(4) 多个场景下的泛化测试和验证（多线程、perstream queue、ACLgraph和训练混用、确定性计算、taskqueue=0、1/2...）,对于验证无法支持的场景，明确做限制和说明

2.方案设计

2.1 总体方案

# 设置模式
torch.npu.set_task_queue_enable(0)   # 关闭 task queue
torch.npu.set_task_queue_enable(1)   # Level 1 优化 (默认)
torch.npu.set_task_queue_enable(2)   # Level 2 优化 (纯推理)

# 查询当前模式
mode = torch.npu.get_task_queue_enable()  # 返回 0/1/2

2.2技术选型

                      ┌──────────────────────────┐
                      │      Python 层             │
                      │                           │
                      │  set_task_queue_enable()  │
                      │  get_task_queue_enable()  │
                      └──────────┬───────────────┘
                                 │ pybind11
                      ┌──────────▼───────────────┐
                      │      Module.cpp           │
                      │                           │
                      │  _npu_set_task_queue_enable()
                      │    ├─ emptyAllNPUStream() │
                      │    └─ SetTaskQueueEnable()│
                      │                           │
                      │  _npu_get_task_queue_enable()
                      │    └─ GetTaskQueueEnable()│
                      └──────────┬───────────────┘
                                 │
                      ┌──────────▼───────────────┐
                      │   OptionsManager          │
                      │                           │
                      │  g_task_queue_enable_mode │
                      │  (std::atomic<int32_t>)   │
                      │                           │
                      │  GetTaskQueueEnable()     │
                      │    ├─ CheckBlocking()→0   │
                      │    ├─ dynamic != -1 → val │
                      │    └─ env var fallback    │
                      │                           │
                      │  SetTaskQueueEnable(mode) │
                      │    ├─ validate mode       │
                      │    └─ atomic.store(mode)  │
                      └───────────────────────────┘
                                 │
                    ┌────────────┼────────────┐
                    ▼            ▼            ▼
              NPUGraph.cpp  NPUQueue.cpp  Module.cpp
              (const,每次    (const,每次    (const,每次
               重新读取)     重新读取)      重新读取)

3.测试方案

一、TestTaskQueueEnableEnv（功能测试）

#	测试方法	验证内容
1	`test_blocking_override`	`ASCEND_LAUNCH_BLOCKING=1` 时强制 mode 为 0（非该环境自动跳过）
2	`test_invalid_mode`	非法值 `-1` / `3` / `-2` 抛出 `RuntimeError`
3	`test_mode_roundtrip_and_switch`	模式 0/1/2 设置后读取值一致
4	`test_npu_graph_capture_rejected_at_mode2`	mode=2 时 NPU Graph 捕获被拒绝（抛 `RuntimeError`）
5	`test_per_stream_queue_interaction_mode0`	子进程中 `PER_STREAM_QUEUE=1` + 动态切换 mode=0 运行正常
6	`test_deterministic_level_not_reset_by_mode_switch`	模式切换不会重置 deterministic level

公共机制

setUp：保存当前 task queue 模式
tearDown：恢复原始模式，保证测试隔离

二、TestGetTaskQueueEnableTiming（性能测试）

#	测试方法	验证内容
1	`test_get_task_queue_enable_timing_all_paths`	get 接口各路径冒烟测试
2	`test_get_task_queue_enable_call_latency`	set=1 fast path 与 set=2 路径的调用延迟对比（ns/call）
3	`test_multithread_get_dispatch`	1/2/4 个 reader 并发 get + writer 切换模式：读值合法、无异常，并统计吞吐