智能家居多设备AI推理优先级调度方案：紧急事件抢占与周期性任务的混合实时调度设计

智能家居多设备AI推理优先级调度方案：紧急事件抢占与周期性任务的混合实时调度设计

一、深度引言

智能家居边缘网关需要同时运行多个AI推理任务：人脸识别（门锁联动）、语音唤醒（音箱交互）、手势识别（面板控制）、异常行为检测（安防摄像头）。这些任务在单个嵌入式AI加速器（如1 TOPS NPU）上竞争算力资源时，简单的FIFO调度会导致安防告警被语音命令阻塞、关键事件延迟飙升到不可接受的范围。

本文将设计一套基于固定优先级抢占式调度（Fixed-Priority Preemptive Scheduling, FPPS）的混合实时调度方案，将AI推理任务按紧急程度分为三个优先级层级：紧急事件级（火灾检测、入侵告警，延迟要求<200 ms）、交互响应级（语音控制、手势识别，延迟要求<500 ms）、后台分析级（能耗优化、用户行为分析，延迟要求<2 s）。通过时间分割复用 + 推理中断恢复机制，在单NPU硬件上实现多任务的安全共存。

硬件平台：Rockchip RK3568（4核Cortex-A55 + 1 TOPS NPU），1 GB RAM，Linux 5.10 + RT-Preempt补丁。

二、原理剖析

2.1 任务优先级建模

各任务特性定义：

• T（周期）：任务的触发间隔，由传感器数据到达频率决定。
• C（计算时间）：NPU上完成一次推理的最坏情况执行时间（WCET），通过1000次实测的最大值加20%安全余量得出。
• D（截止时间）：任务必须完成的最坏允许延迟，通常等于T（隐式截止时间）。

2.2 可调度性分析

对于固定优先级抢占式调度，采用响应时间分析（Response Time Analysis, RTA）验证可调度性。任务τ_i的最坏响应时间R_i通过以下迭代公式计算：

R_i^(0) = C_i
R_i^(n+1) = C_i + Σ_{j∈hp(i)} ceil(R_i^(n) / T_j) × C_j

其中hp(i)为优先级高于τ_i的任务集合。当R_i^(n+1) = R_i^(n)或R_i^(n) > D_i时迭代终止。若R_i ≤ D_i对所有任务成立，则任务集可调度。

以本文任务集为例（C_i单位ms）：

P3（KWS, T=50ms）无法通过RTA分析，因为高优先级任务P0~P2的中断累积使P3的最坏响应时间（127 ms）超过其截止时间（50 ms）。解决方案：引入推理分片机制，允许长时间推理任务被抢占后恢复。

2.3 推理分片与检查点恢复

推理分片（Inference Slicing）的核心思想：将深度神经网络的计算分解为可中断的层间边界。当高优先级任务到达时，NPU在当前层完成后立即保存上下文（层索引 + 活跃Tensor的DMA地址），切换到高优先级任务，完成后恢复下层上下文并继续推理。层间切换延迟约50~100 μs（取决于Tensor转储大小），远低于完整的推理等待时间。

三、代码实现

/**
* @file npu_rt_scheduler.c
* @brief NPU实时推理调度器 – 固定优先级抢占 + 推理分片恢复
* @note 平台: RK3568 (4×Cortex-A55 + 1TOPS NPU)
* OS: Linux 5.10 + RT-Preempt
* NPU驱动: RKNN v1.5+
* 设计原则:
* 1. 固定优先级: 紧急 > 交互 > 后台
* 2. 抢占式: 高优先级任务可中断低优先级推理
* 3. 分片恢复: 利用NPU层间边界实现安全抢占
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdatomic.h>
#include <errno.h>
#include <time.h>
#include <rknn_api.h>

/* ———- 任务优先级定义 ———- */
typedef enum {
PRIO_EMERGENCY = 0, /* 紧急事件: 火灾/入侵/跌倒 */
PRIO_INTERACTIVE = 1, /* 交互响应: 语音/人脸/手势 */
PRIO_BACKGROUND = 2, /* 后台分析: 能耗/行为/预测 */
PRIO_NUM_LEVELS = 3
} task_priority_t;

#define MAX_TASKS 16
#define MAX_CHECKPOINTS 32 /* 每个模型最大保存点数 */
#define INFER_TIMEOUT_MS 5000 /* 推理超时 */

/* ———- 推理检查点 (用于抢占后恢复) ———- */
/**
* @brief 推理检查点 – 保存模型推理的中间状态
* @note 检查点设置在每个layer/layer group之后
* 恢复时从最近一个检查点继续, 而非从头开始
*/
typedef struct {
uint32_t layer_index; /* 已完成的最后一层索引 */
uint32_t next_layer_index; /* 恢复时继续执行的第一层 */
uint64_t intermediate_addr; /* 中间Tensor在NPU内存中的DMA地址 */
size_t intermediate_size; /* 中间Tensor大小 (字节) */
uint32_t checksum; /* 中间状态校验和 (完整性验证) */
} infer_checkpoint_t;

/* ———- 推理任务描述 ———- */
typedef struct infer_task {
int task_id;
task_priority_t priority;
uint64_t period_us; /* 触发周期 (us) */
uint64_t deadline_us; /* 截止时间 (us) */
uint64_t wcet_us; /* 最坏执行时间 (us) */
rknn_context ctx; /* RKNN模型上下文 */
int model_id; /* 模型唯一标识 */
atomic_bool running; /* 推理进行中标志 */
infer_checkpoint_t checkpoints[MAX_CHECKPOINTS]; /* 检查点数组 */
int num_checkpoints; /* 检查点数量 */

/* 统计信息 */
uint64_t total_invocations;
uint64_t total_preemptions; /* 被抢占次数 */
uint64_t max_response_us; /* 最大响应时间 */
uint64_t deadline_misses; /* 截止时间超时次数 */
} infer_task_t;

/* ———- NPU全局锁与调度状态 ———- */
static pthread_mutex_t g_npu_mutex = PTHREAD_MUTEX_INITIALIZER;
static atomic_int g_npu_owner_task_id = -1; /* 当前占用NPU的任务 */
static atomic_bool g_preemption_requested = false; /* 抢占请求标志 */
static infer_task_t *g_task_list[MAX_TASKS];
static int g_num_tasks = 0;

/* ———- 抢占请求接口 ———- */
/**
* @brief 请求NPU抢占 (由高优先级任务调用)
* @param requester 发起抢占请求的任务
* @return 0抢占成功, -1当前无任务占用NPU, -2抢占超时
* @note 实现:
* 1. 设置全局抢占标志位
* 2. 等待当前任务释放NPU (最大等待100ms)
* 3. 获取NPU mutex → 成为NPU新占有者
*/
int npu_request_preemption(infer_task_t *requester)
{
struct timespec timeout;
int ret;

if (!requester) return -1;

/* 设置抢占标志 */
atomic_store(&g_preemption_requested, true);

/* 等待NPU可用 (带超时) */
clock_gettime(CLOCK_MONOTONIC, &timeout);
timeout.tv_nsec += 100000000; /* +100ms */
if (timeout.tv_nsec >= 1000000000) {
timeout.tv_sec += 1;
timeout.tv_nsec -= 1000000000;
}

ret = pthread_mutex_timedlock(&g_npu_mutex, &timeout);
if (ret == ETIMEDOUT) {
printf("[SCHED] 抢占NPU超时(100ms), requester=T%d prio=%d\\n",
requester->task_id, requester->priority);
atomic_store(&g_preemption_requested, false);
return -2; /* 超时: 当前任务可能卡死, 需触发硬件复位 */
}

if (ret != 0) {
printf("[SCHED] NPU mutex获取失败: %d\\n", ret);
atomic_store(&g_preemption_requested, false);
return -1;
}

/* 抢占成功, 记录新NPU持有者 */
atomic_store(&g_npu_owner_task_id, requester->task_id);
atomic_store(&g_preemption_requested, false);

return 0;
}

/* ———- 检查点保存 ———- */
/**
* @brief 保存推理检查点
* @param task 推理任务
* @param layer_idx 当前完成的层索引
* @note 在每层推理完成后调用
* 当检测到抢占请求时, 保存状态后释放NPU
*/
static int save_checkpoint(infer_task_t *task, int layer_idx)
{
if (!task || layer_idx >= MAX_CHECKPOINTS) {
return -1;
}

infer_checkpoint_t *cp = &task->checkpoints[layer_idx];
cp->layer_index = layer_idx;
cp->next_layer_index = layer_idx + 1;

/* 从RKNN获取中间Tensor信息 */
rknn_tensor_mem *intermediate = NULL;
/* int ret = rknn_query(task->ctx, RKNN_QUERY_INTERMEDIATE,
intermediate, sizeof(*intermediate)); */
/* 若查询成功:
cp->intermediate_addr = intermediate->phys_addr;
cp->intermediate_size = intermediate->size;
*/

/* 计算校验和 (简化: layer_index × 0x5A5A) */
cp->checksum = layer_idx * 0x5A5A5A5A;

if (layer_idx >= task->num_checkpoints) {
task->num_checkpoints = layer_idx + 1;
}

return 0;
}

/* ———- 推理分片执行 (带抢占感知) ———- */
/**
* @brief 分片推理执行函数
* @param task 推理任务
* @param start_layer 起始层索引 (0=从头开始, >0=从检查点恢复)
* @return 0完成全部推理, -1被抢占, -2推理异常
* @note 每完成一层后检查抢占标志, 若置位则保存检查点并释放NPU
* 恢复时从最近检查点继续, 避免重复计算
*/
static int execute_inference_sliced(infer_task_t *task, int start_layer)
{
int total_layers = 10; /* 假设模型共10层, 实际应从rknn_query获取 */
int current_layer = start_layer;
int ret;

for (; current_layer < total_layers; current_layer++) {
/* 执行单层推理 */
/* ret = rknn_run_layer(task->ctx, current_layer); */
/* 模拟: usleep(task->wcet_us / total_layers); */

/* 保存检查点 (为抢占做准备) */
ret = save_checkpoint(task, current_layer);
if (ret != 0) {
printf("[SCHED] T%d: 层%d检查点保存失败\\n",
task->task_id, current_layer);
return -2;
}

/* 检查抢占请求 */
if (atomic_load(&g_preemption_requested)) {
printf("[SCHED] T%d: 在层%d检测到抢占请求, 释放NPU\\n",
task->task_id, current_layer);

/* 记录抢占统计 */
task->total_preemptions++;

/* 释放NPU mutex (允许高优先级任务获取) */
atomic_store(&g_npu_owner_task_id, -1);
pthread_mutex_unlock(&g_npu_mutex);

return -1; /* 被抢占, 调用者负责后续恢复 */
}
}

/* 所有层完成 → 最终后处理 */
/* ret = rknn_get_outputs(task->ctx); */

/* 清空检查点 (推理完整结束) */
task->num_checkpoints = 0;

return 0;
}

/* ———- 推理任务主循环 ———- */
/**
* @brief 推理任务主循环
* @param arg 指向infer_task_t的指针
* @note 每个推理任务独立线程运行
* 执行流程:
* 1. 等待触发信号 (传感器数据就绪)
* 2. 记录时间戳 (用于响应时间统计)
* 3. 根据优先级获取NPU (可能触发抢占)
* 4. 执行推理或从检查点恢复
* 5. 释放NPU, 检查截止时间是否超时
*/
static void *infer_task_loop(void *arg)
{
infer_task_t *task = (infer_task_t *)arg;

if (!task) {
return NULL;
}

struct timespec trigger_time, complete_time;

while (1) {
/* 等待触发 (简化: 定时周期, 实际由传感器事件驱动) */
usleep(task->period_us);

/* 记录触发时间 */
clock_gettime(CLOCK_MONOTONIC, &trigger_time);
task->total_invocations++;

int start_layer = 0;
int infer_result;

/* 循环直到推理完成 (可能经历多次抢占-恢复) */
do {
/* 根据优先级策略获取NPU */
if (task->priority == PRIO_EMERGENCY) {
/* 紧急任务: 立即抢占 */
int preempt_ret = npu_request_preemption(task);
if (preempt_ret == -2) {
/* 抢占超时: 触发硬件看门狗复位NPU */
/* nnp_reset_hardware(); */
printf("[SCHED] T%d: NPU抢占超时, 触发硬件复位\\n",
task->task_id);
continue; /* 下一周期重试 */
}
} else {
/* 非紧急任务: 等待NPU空闲或被动被唤醒 */
int lock_ret = pthread_mutex_lock(&g_npu_mutex);
if (lock_ret != 0) {
continue;
}
atomic_store(&g_npu_owner_task_id, task->task_id);
}

/* 执行分片推理 (含抢占感知) */
infer_result = execute_inference_sliced(task, start_layer);

if (infer_result == -1) {
/* 被抢占: 记录检查点位置, 等待重新获得NPU */
if (task->num_checkpoints > 0) {
start_layer = task->checkpoints[
task->num_checkpoints – 1].next_layer_index;
}
/* 循环重新获取NPU并继续 */
continue;
} else {
/* 完成或异常 */
atomic_store(&g_npu_owner_task_id, -1);
pthread_mutex_unlock(&g_npu_mutex);
break;
}
} while (infer_result == -1);

/* 记录完成时间 */
clock_gettime(CLOCK_MONOTONIC, &complete_time);

/* 计算响应时间 */
uint64_t response_us =
(complete_time.tv_sec – trigger_time.tv_sec) * 1000000ULL +
(complete_time.tv_nsec – trigger_time.tv_nsec) / 1000;

/* 更新最大响应时间 */
if (response_us > task->max_response_us) {
task->max_response_us = response_us;
}

/* 检查截止时间 */
if (response_us > task->deadline_us) {
task->deadline_misses++;
printf("[SCHED] T%d: 截止时间超时! response=%lluus > deadline=%lluus (miss #%llu)\\n",
task->task_id,
(unsigned long long)response_us,
(unsigned long long)task->deadline_us,
(unsigned long long)task->deadline_misses);
}
}

return NULL;
}

/* ———- 调度器初始化 ———- */
/**
* @brief 注册推理任务到调度器
* @param task 推理任务描述
* @return 0成功, -1参数错误, -2任务表满
*/
int sched_register_task(infer_task_t *task)
{
if (!task || task->priority >= PRIO_NUM_LEVELS) {
return -1;
}

if (g_num_tasks >= MAX_TASKS) {
printf("[SCHED] 任务表满 (max=%d)\\n", MAX_TASKS);
return -2;
}

task->task_id = g_num_tasks;
g_task_list[g_num_tasks] = task;
g_num_tasks++;

/* 创建任务线程 */
pthread_t thread;
pthread_attr_t attr;
pthread_attr_init(&attr);

/* 设置调度策略: SCHED_FIFO (实时) */
struct sched_param sched_param;
/* 优先级映射: 紧急=90, 交互=60, 后台=30 */
sched_param.sched_priority = 90 – task->priority * 30;
pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
pthread_attr_setschedparam(&attr, &sched_param);
pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);

pthread_create(&thread, &attr, infer_task_loop, task);
pthread_attr_destroy(&attr);

printf("[SCHED] 注册任务 T%d, prio=%d, period=%lluus\\n",
task->task_id, task->priority,
(unsigned long long)task->period_us);

return 0;
}

/* ———- 统计报告 ———- */
void sched_print_statistics(void)
{
printf("\\n========== NPU调度器统计报告 ==========\\n");
printf("%-6s %-8s %-12s %-12s %-12s %-12s\\n",
"Task", "Prio", "调用次数", "抢占次数", "最大响应(us)", "超时次数");
printf("—————————————–\\n");

for (int i = 0; i < g_num_tasks; i++) {
infer_task_t *t = g_task_list[i];
if (!t) continue;
printf("T%-5d %-8d %-12llu %-12llu %-12llu %-12llu\\n",
t->task_id, t->priority,
(unsigned long long)t->total_invocations,
(unsigned long long)t->total_preemptions,
(unsigned long long)t->max_response_us,
(unsigned long long)t->deadline_misses);
}
printf("=========================================\\n");
}

四、边界分析

4.1 抢占风暴与优先级反转

当多个紧急任务几乎同时到达时（如火灾检测和入侵检测在50 ms内相继触发），可能产生抢占风暴——任务A抢占B、B恢复后A再次到达又抢占B，导致后台任务饥饿。本方案通过设置"紧急任务冷却期"缓解：同一紧急任务在完成后的200 ms内不重复触发（通过时间戳比较过滤），防止传感器噪声导致的频繁误触抢占。

优先级反转的典型场景：低优先级任务持有NPU mutex时被中优先级任务等待，高优先级任务又试图获取mutex——此时mutex被低优先级持有，高优先级被阻塞。解决方案：使用PTHREAD_PRIO_INHERIT互斥锁属性，使持有mutex的低优先级任务临时继承等待者的最高优先级，防止中优先级任务的干扰。

4.2 检查点的一致性边界

检查点恢复的前提是：上一层的输出Tensor在抢占期间未被修改。这要求NPU的中间Tensor存储区在任务切换时保持隔离。若NPU驱动不支持多上下文中间Tensor隔离（即所有任务共享同一块中间存储区），则抢占后的恢复无法从检查点继续，必须从第一层重新推理。这会使抢占开销从层切换延迟（~100 μs）变为完整推理的重计算（~数十ms）。在RK3568的RKNN v1.5中，每个rknn_context拥有独立的中间Tensor内存空间，支持安全的上下文切换。

4.3 调度器自身的CPU开销

调度器运行在Cortex-A55上（非NPU），其CPU开销需要被计入任务的计算时间预算。关键开销项：

对于周期50 ms的KWS任务，调度器开销约150 μs/周期（0.3% CPU），可忽略。但对于周期10 ms的高频传感器任务，需考虑调度器CPU占用率。

4.4 NPU驱动的抢占支持边界

当前大多数边缘NPU（包括RKNN、T-engine、Himax WE-I Plus等）不支持硬件级任务抢占——即不能在单条指令执行过程中暂停NPU流水线。本文的方案通过在层间边界（软件级）实现抢占，利用的是"层间同步点"这一天然边界。局限性在于：如果某层的计算时间过长（如Transformer的Self-Attention矩阵乘），任务将在整个层执行期间不可抢占，可能违反紧急任务的截止时间。对于超大层模型，应在模型转换阶段插入"层内分割点"（通过算子拆分，如将大矩阵乘分解为多个小矩阵乘），以增加抢占机会。

五、总结

本文设计了面向智能家居边缘网关的多设备AI推理优先级调度方案。核心结论：