# Calculet NPU 异步执行器与生产稳定性实施规格

日期：2026-08-02  
适用基线：CalRT 0.7.6、llama.cpp `fd9bd632`、当前 Qwen3 batch=1 双芯粒 calbin

## 1. 文档目的和边界

本文不是架构草图，而是改造当前 `llama-calrt.cpp` 的实施规格。第一阶段只做 C2 工程改造，不假设 Runtime 支持多物理设备，也不把未验证的 parallel/ping-pong 作为生产能力。

完成后应得到：

- `infer()`、CSR、`Wait()` 的所有错误都能向上返回，不再 `abort()`。
- KV 不足有界等待，不再 CPU 无限忙等。
- 每个在途任务独占 InputBuf、OutputBuf 和所有 host backing storage。
- 请求线程、设备提交线程、完成线程解耦；Phase A 仍保持 `max_inflight=1`。
- 取消、超时、设备异常、模型切换都有可验证的状态机。
- Phase B 才在隔离 runner 中验证 ping/pong；验证不通过可无代码回退到 1 in-flight。

## 2. 当前代码级问题清单

| ID | 当前实现 | 直接风险 | 第一落点 |
| --- | --- | --- | --- |
| S01 | `KvManager::Apply()` 失败后 `do...while(!ok)` | 无 sleep、无 deadline、无法取消，可能占满 CPU 永久挂起 | admission queue |
| S02 | infer 后二次 `Apply()` 失败 `abort()` | 单请求资源问题杀死整个服务 | KV transaction + error return |
| S03 | `SetCsrByName()` 返回码未检查 | CSR 名称或值错误仍提交图 | `prepare_job()` |
| S04 | `infer()` 返回码未检查 | submit 失败仍进入 `Wait()` | `submit_job()` |
| S05 | `Wait()` 返回码未检查 | timeout/CCU exception 被当作成功 | `complete_job()` |
| S06 | context 只有一组 prefill/decode vectors | 第二任务可覆盖第一任务 DMA 内存 | `BufferLease` 独占 backing |
| S07 | 每次创建 Runtime buffer | 热路径分配和初始化开销 | generation-scoped pool |
| S08 | substring 选择子模型且最后匹配覆盖前者 | 名称碰撞时选错图 | exact manifest |
| S09 | prefill output `D0=16` | 新布局下 slice/untile 错误 | layout descriptor |
| S10 | 整个 BF16 vocab 先转临时 FP32 vector | 多一次分配、遍历和拷贝 | 直接转换到目标区 |
| S11 | 计时存在 `std::map` 且按 seq 共享 | 并发读写数据竞争、历史值残留 | per-job timings + metrics |
| S12 | dump 路径硬编码用户目录 | 隐私、磁盘和可移植性风险 | 默认禁用的受控 dump sink |

## 3. 代码改动边界

建议保持 llama.cpp 上层调度接口稳定，新建四个文件并收缩原有文件职责：

```text
src/llama-calrt.cpp                 只保留 llama batch 适配和同步兼容入口
src/llama-calrt.h                   对上接口、manifest 和结果类型
src/llama-calrt-executor.h/.cpp     DeviceExecutor、JobContext、队列和 completion
src/llama-calrt-buffer-pool.h/.cpp  BufferPool、BufferLease、layout 校验
src/llama-calrt-kv-txn.h/.cpp       KV reservation/commit/rollback
src/llama-calrt-error.h/.cpp        CalRT 错误分类、日志和服务错误映射
```

`llama-kv-cache-calrt-adapter.cpp` 只声明已经真实支持的语义。空实现不能继续被当成成功；无法实现的 `seq_cp/keep/add/div/state_*` 应显式返回“不支持”或在调用前禁用对应上层功能。

## 4. 精确数据类型

### 4.1 模型清单

```cpp
enum class CalTaskKind : uint8_t { PrefillIds, PrefillEmbeds, Decode, Vision };

struct TensorContract {
    std::string name;
    std::vector<uint64_t> shape;
    calrt::PrimitiveType dtype;
    uint64_t byte_size;
    bool sliceable;
};

struct SubmodelContract {
    CalTaskKind kind;
    std::string exact_name;
    TensorContract token_or_embed;
    TensorContract position;
    TensorContract logits;
    std::string csr_cur_seq;
    std::string csr_total_seq;
    uint32_t max_batch;
    uint32_t max_seq;
    uint32_t logits_tile_d0;
};

struct ModelGeneration {
    uint64_t generation_id;
    std::string calbin_sha256;
    std::string calrt_version;
    std::unordered_map<CalTaskKind, SubmodelContract> submodels;
};
```

当前产物 manifest 的固定值是：prefill/decode `max_batch=1`、`max_seq=40960`、token/position 为 S32、logits 为 BF16 `[1,1,151936]`。`logits_tile_d0=16` 只能由当前图 metadata 解析后写入 manifest，不能作为代码默认值。

### 4.2 任务与 buffer

```cpp
enum class JobState : uint8_t {
    Created, Admitted, Preparing, Queued, Submitted,
    Completing, Succeeded, Failed, CancelRequested, Quarantined
};

enum class EngineSlot : int8_t { Auto = -1, Ping = 0, Pong = 1 };

struct JobTimings {
    int64_t accepted_us = 0;
    int64_t admitted_us = 0;
    int64_t queued_us = 0;
    int64_t submitted_us = 0;
    int64_t completed_us = 0;
    double h2d_ms = 0;
    double device_wait_ms = 0;
    double d2h_ms = 0;
    double postprocess_ms = 0;
};

struct BufferStorage {
    std::vector<uint8_t> input0;
    std::vector<uint8_t> position;
    std::vector<uint8_t> output0;
};

struct BufferSlot {
    uint64_t generation_id;
    CalTaskKind task;
    EngineSlot engine;
    std::unique_ptr<calrt::CalrtInputBuf> input;
    std::unique_ptr<calrt::CalrtOutputBuf> output;
    BufferStorage host;
    bool sliced = false;
    bool quarantined = false;
};

struct JobContext {
    uint64_t job_id;
    uint64_t generation_id;
    uint64_t request_id;
    calrt::KvSeqId seq_id;
    CalTaskKind task;
    EngineSlot engine;
    uint32_t past_len;
    uint32_t token_count;
    std::chrono::steady_clock::time_point deadline;
    std::atomic<JobState> state{JobState::Created};
    std::atomic<bool> cancel_requested{false};
    std::optional<BufferLease> buffers;
    std::optional<KvReservation> kv;
    JobTimings timings;
    Status result;
    std::promise<InferenceResult> promise;
};
```

`BufferLease` 必须是 move-only。它的析构只能在 `Succeeded/Failed` 且确认 Runtime 不再 DMA 时把 slot 归还池；`Submitted` 状态析构应触发断言并转入 quarantine，而不是静默复用。

## 5. 所有权与锁不变量

### 5.1 强制不变量

1. `DeviceExecutor` 独占 `VirtualDevice`、`Calbin` 和 `KvManager`。
2. 只有 submitter 线程调用 `infer()`；只有 completion 线程对已提交 job 调用 `Wait()`。
3. 一个 `BufferSlot` 同一时刻最多属于一个 `JobContext`。
4. `MapBuf()` 后 host vector 在完成前不得 resize、move、swap 或释放。
5. job 的 `generation_id` 必须与 pool/model generation 一致。
6. KV reserve 成功不等于 KV commit；只有设备完成且输出通过校验才 commit。
7. cancel 不释放底层资源，只阻止结果交付；仍需完成 `Wait()` 或执行已验证的 reset。
8. reset 后所有旧 generation buffer 一律 quarantine 并重建。

### 5.2 锁和顺序

| 锁 | 保护对象 | 可持锁调用 Runtime 吗 |
| --- | --- | --- |
| `lifecycle_mu` | model generation、Ready/Draining/Recovering | configure/reset 可在独占阶段调用；不能同时持 queue 锁 |
| `admission_mu` | KV 等待队列、token budget | 否 |
| `pool_mu` | free/leased/quarantine 集合 | 否 |
| `submit_mu` | submit queue 条件变量 | 否 |
| `completion_mu` | submitted job 队列 | `Wait()` 前必须释放 |

若一次操作需要多个锁，固定顺序为：`lifecycle_mu -> admission_mu -> pool_mu -> submit_mu`。completion 线程不可持任何上述锁执行阻塞 `Wait()`。

## 6. 端到端调用流

### 6.1 同步兼容入口

```cpp
StatusOr<InferenceResult> calrt_context::decode_sync(const llama_batch &batch) {
    auto request = build_request(batch);
    CAL_ASSIGN_OR_RETURN(auto handle, executor_->enqueue(std::move(request)));
    return handle.wait_until(request.deadline);
}
```

原 `int calrt_decode(...)` 可暂时把 `Status` 映射成 llama 返回码，但日志必须保留 `job_id`、`seq_id`、generation、CalRT 原始错误码和阶段。

### 6.2 `enqueue()`

```text
validate request
  -> exact submodel lookup
  -> deadline/cancel check
  -> admission reserve KV + token budget
  -> acquire BufferLease
  -> copy inputs + set slices + set CSR
  -> push submit queue
  -> return JobHandle
```

任一步失败都必须按逆序释放已取得资源。尚未调用 `infer()` 时 buffer 可正常归还；已经提交后只能由 completion/recovery 路径处理。

### 6.3 `prepare_job()` 检查顺序

1. `token_count > 0`。
2. `past_len + token_count <= max_seq`，使用 64 位加法防溢出。
3. 当前 calbin batch=1 时，`seq_count == 1`。
4. token 与 position 指针非空，长度一致；embed 路径单独检查元素数。
5. tensor exact name 存在，dtype、rank、shape 和 ByteSize 与 manifest 一致。
6. `input_bytes <= tensor.ByteSize()`；slice offset/size 均在范围内。
7. 两个 `SetCsrByName()` 都返回 `CalrtSuccess`。
8. `MapBuf()` 和 `SliceTensor()` 的返回/状态均检查。

### 6.4 submitter

```cpp
void DeviceExecutor::submit_one(std::shared_ptr<JobContext> job) {
    if (job->cancel_requested.load()) {
        fail_before_submit(job, Status::Cancelled());
        return;
    }

    const auto rc = use_fixed_engine_
        ? calrt::infer_with_fixed_task_type(vdev_.get(), model(job),
              job->buffers->input(), job->buffers->output(), to_task_type(job->engine))
        : calrt::infer(vdev_.get(), model(job),
              job->buffers->input(), job->buffers->output());

    if (rc != CalrtSuccess) {
        rollback_before_device_execution(job, from_calrt(rc, "submit"));
        return;
    }
    transition(job, JobState::Queued, JobState::Submitted);
    completion_queue_.push(job);
}
```

Phase A 的 semaphore 容量固定为 1。Phase B 只有 runner 验证后才改成 2，并给每个 slot 固定 Ping/Pong；不能把并发 HTTP 请求数直接当作 in-flight 上限。

### 6.5 completion

```text
pop submitted job (不持锁)
  -> OutputBuf::Wait()
  -> 读取 OutputBuf::GetStatus()
  -> rc/status 双重判定
  -> 检查 tensor shape/size 和 finite logits
  -> 直接 BF16 -> 目标 FP32 区
  -> KV commit
  -> reset slice/tensors
  -> promise success
  -> release BufferLease
```

若 `Wait()` 返回失败、状态为 `DONE_CCU_EXCEPTION` 或输出校验失败：不 commit KV；相关 sequence 标记 Unknown；buffer quarantine；触发故障控制器判定是否 drain/reset。

## 7. KV 事务实现

```cpp
enum class KvTxnState : uint8_t { Reserved, Submitted, Committed, RolledBack, Unknown };

struct KvReservation {
    calrt::KvSeqId seq_id;
    uint32_t old_len;
    uint32_t target_len;
    uint64_t generation_id;
    KvTxnState state;
};

class KvCoordinator {
  public:
    StatusOr<KvReservation> reserve(KvSeqId id, uint32_t target,
                                    std::chrono::steady_clock::time_point deadline);
    Status mark_submitted(KvReservation &txn);
    Status commit(KvReservation &txn);
    Status rollback(KvReservation &txn);
    void mark_unknown(KvReservation &txn, std::string_view reason);
};
```

0.7.6 的 `KvManager::Apply()` 返回 bool 而无错误原因。短期规则：

- reserve 前先用 `canAllocate(1)`；false 时进入条件变量等待，而非轮询。
- 每次唤醒只重试一次 `Apply()`，重新检查 deadline 和 shutdown。
- submit 前 Apply 的目标长度应由现有语义 runner 固化，不能同时改变长度口径与异步架构。
- submit 失败后若 Runtime 没执行图，可将软件 transaction rollback；硬件 KV 是否已经变化必须通过 vendor 契约或 golden 证明。
- submit 成功后任何错误默认 `Unknown`，不能声称请求级 rollback 已完成。

## 8. 错误分类和服务动作

| CalRT 错误 | 类别 | 当前 job | 新 admission | 自动动作 |
| --- | --- | --- | --- | --- |
| 1 内存分配、14 越界、21 dtype、22 shape | Request/Artifact | 失败 | 同模型可继续或隔离产物 | 记录 manifest 差异；不 reset |
| 2 已注册、7 已配置、16 缺配置 | Lifecycle | 失败 | 暂停 | drain，核对 generation/config |
| 4 非法值、5 非法配置、19 非法 calbin、10/11/12 文件 | Artifact | 失败 | 拒绝该 generation | 回滚上个已知好模型 |
| 8 driver 不兼容、9 RT 过旧、23 依赖错误 | Compatibility | 失败 | 停止 | 人工修复，不自动重试 |
| 13 timeout、17 busy | Transient/Unknown | 失败或重试前排队 | 降低 admission | 有界退避；达到阈值 drain |
| 18 unavailable、24 map broken、25 soft reset 失败、28 crash | Device | 失败 | 立即停止 | quarantine -> reset/restart/escalate |
| 27 不支持、29 方向错误 | Programming | 失败 | 相关功能关闭 | 不 reset，修代码/配置 |
| 15/20 未预期/未知 | Unknown | 失败 | 暂停 | 保守按 device generation 污染处理 |

推荐 HTTP/gRPC 映射：输入越界 `INVALID_ARGUMENT/400`；KV/队列满 `RESOURCE_EXHAUSTED/429`；deadline `DEADLINE_EXCEEDED/504`；设备恢复中 `UNAVAILABLE/503`；内部 shape/calbin 错误 `INTERNAL/500` 并摘除 generation。

## 9. 超时、取消与恢复

### 9.1 三类 deadline

| deadline | 开始 | 结束 | 超时动作 |
| --- | --- | --- | --- |
| admission | 请求进入服务 | KV+buffer 取得 | 从等待队列移除，不碰设备 |
| queue | job prepared | `infer()` 成功 | 未提交可直接 rollback |
| execution | `infer()` 成功 | `Wait()` 返回 | 不释放 DMA buffer，进入 Recovering |

0.7.6 没有已确认的 job cancel API。execution timeout 后不能 detach completion 再复用资源；默认步骤为停止 admission、保留所有 submitted buffer、等待短 grace period，然后走厂商确认过的 reset。没有 reset 契约前，最安全的最终恢复手段是进程退出，由 supervisor 重启。

### 9.2 设备恢复状态机

```text
Ready -> Draining -> Recovering -> Configuring -> Warming -> Ready
  |          |             |             |
  +--------> Failed <------+-------------+
```

恢复需要 generation+1，重新创建 Calbin/KvManager/buffer pool，跑固定 prefill+decode golden，成功后才接流。旧 completion 只能写旧 job promise，不得更新新 generation 的 KV 或指标 map。

## 10. 直接优化项

### 10.1 logits 转换

把：

```cpp
std::vector<float> src(element_num);
for (...) src[i] = ggml_bf16_to_fp32(src_bf16[i]);
memcpy(dst, src.data() + base, dict_len * sizeof(float));
```

改为仅转换需要的 vocab 区：

```cpp
for (size_t i = 0; i < dict_len; ++i) {
    dst[i] = ggml_bf16_to_fp32(src_bf16[base + i]);
}
```

边界检查使用 `base <= tensor_size && dict_len <= tensor_size - base`，避免 `base + dict_len` 溢出。后续再基于 profiling 选择 SIMD 或 NPU top-k；后者是 C3。

### 10.2 exact submodel lookup

启动时枚举一次 `GetAllModels()`，验证 manifest 中每个 exact name 恰好出现一次。禁止运行时 substring fallback。缺失、多余或同 kind 重复都阻止 model generation 进入 Ready。

### 10.3 metrics guard

`server.cpp` 所有 `cal_ctx->get_*` 必须完全置于 `#ifdef USE_CALRT` 内。计时从共享 `std::map<seq_id,double>` 改到 `InferenceResult.timings`，请求结束即聚合到直方图，避免 seq id 复用污染。

## 11. 分阶段提交计划

| PR | 代码范围 | 行为变化 | 必须测试 |
| --- | --- | --- | --- |
| PR1 | error wrapper、所有返回码、metrics guard | 错误可见，不改调度 | 错误映射单测、现有 golden |
| PR2 | manifest exact lookup、tensor/layout 校验 | 启动 fail-fast | 缺模型/重名/shape/dtype 注入 |
| PR3 | KV 有界 admission、移除 abort | 资源不足变排队/错误 | deadline/cancel/容量用尽 |
| PR4 | BufferPool + move-only lease | 减少分配、明确所有权 | 地址稳定、归还重置、quarantine |
| PR5 | JobContext + submit/completion，in-flight=1 | 线程解耦、行为等价 | TSAN、取消、shutdown、1000 次循环 |
| PR6 | logits 直接转换 | CPU/内存优化 | bitwise/误差、边界、perf |
| PR7 | ping/pong runner，默认关闭 | 只形成实验能力 | 双 sequence 串扰、状态、fault |
| PR8 | 产品灰度 `max_inflight=2` | 可能提高 overlap | 24h soak、P99、自动降级 |

每个 PR 都可独立回滚。PR7/PR8 不应与 batch calbin 接入混在同一变更中。

## 12. 故障注入表

| FI | 注入点 | 注入方式 | 预期结果 |
| --- | --- | --- | --- |
| FI01 | exact lookup | manifest 改错一个字符 | 启动失败，未 configure |
| FI02 | CSR | 使用不存在名称 | job prepare 失败，未 submit |
| FI03 | tensor slice | size=ByteSize+1 | 本地越界拒绝，未调用 Runtime |
| FI04 | KV capacity | mock `canAllocate=false` | 等待到 deadline，CPU 不忙等 |
| FI05 | submit | wrapper 返回 `DeviceBusy` | job rollback，无 Wait |
| FI06 | submit | wrapper 返回 `DeviceUnavailable` | admission 停止，进入恢复 |
| FI07 | Wait | 返回 `Timeout` | buffer/KV quarantine，不交付 logits |
| FI08 | status | `DONE_CCU_EXCEPTION` | 即使 Wait=success 也失败并恢复 |
| FI09 | output | tensor size 小于 vocab | 检测失败，无越界写 |
| FI10 | output | 注入 NaN/Inf | generation 隔离，golden 复测 |
| FI11 | cancel before submit | 设置 cancel flag | 无设备提交，资源释放 |
| FI12 | cancel after submit | 完成前取消 | 仍 Wait，结果丢弃，资源最终释放 |
| FI13 | shutdown | 有 1 个 queued + 1 submitted | queued 取消，submitted drain |
| FI14 | stale completion | reset 后返回旧 job | 不写新 generation |
| FI15 | pool reset | 故意不 UndoSlice | 归还检查失败，slot quarantine |

## 13. 验收门槛

### 13.1 正确性

- 固定 100 个 prompt，NPU 改造前后 greedy token 序列完全一致；若参考实现允许浮点误差，另报 top-1 一致率、top-5 overlap 和 logits `max_abs/max_rel/cosine`。
- 输入长度覆盖 1、2、15、16、17、127、128、1024、4095、4096、40959；40960 加 decode 必须被边界检查拒绝。
- 两个 seq id 交替 1000 token 无输出或 KV 串扰。

### 13.2 稳定性

- Phase A 24h，零进程 abort、零无限等待、零未解释 CCU exception。
- RSS、Runtime device memory 和 pool leased 数在 warmup 后无单调增长；终止后 leased=0。
- 取消/超时/错误注入 1000 次后仍可通过 golden，或进入明确 Failed 状态由 supervisor 恢复。

### 13.3 性能

- PR1-PR5 行为重构阶段，单请求 TTFT/TPOT P50 退化不超过 3%，P99 不超过 5%。
- logits 直接转换单独报告 CPU time 和分配次数；不得只看端到端平均值。
- Phase B 只有在 aggregate tokens/s 提升至少 10%、单请求 TPOT P99 退化不超过 5%、正确性和 24h 稳定性均通过时才默认启用。

## 14. 厂商阻断项

以下项不能由当前源码推断：

- `EnableParallelMode(true)` 的线程安全、最大 in-flight 和关闭时 drain 语义。
- PING/PONG 是否真的重叠 H2D/compute/D2H，以及 engine slot 的复用时点。
- execution timeout 后安全 cancel 方法。
- submit 已成功但 Wait 失败时 KV 是否部分更新。
- `ResetCCU`、`ResetConfiguration`、`Reset` 对 DMA、KV、另一芯粒和 host buffer 的影响。

厂商交付必须包含最小 runner、时序图、错误注入结果和恢复后的 golden；口头说明不足以把这些能力从 C1/C3 提升为 C0。