Pipeline
概述
Pipeline（流水线）是一种处理音频流数据的统一框架组件，将音频流处理过程拆解成串行模式，提供了一种可灵活定义数据处理步骤的方法。上层应用可以按照不同的需求，依序预设好数据处理操作即可。
Pipeline框架借鉴了Linux ALSA框架的设计理念，通过流水线方式组织音频处理流程，使开发者能够将复杂的音频处理流程分解为多个独立的处理单元（processor），每个处理单元负责特定的数据处理任务。这种方式具有以下优势：
模块化设计：每个处理单元职责单一，易于开发和维护
灵活组合：可根据不同应用场景自由组合处理单元
低耦合：处理单元之间通过标准接口通信，降低耦合度
高复用性：通用处理单元可在多个stream中复用
统一管理：通过stream统一管理整个处理流程的生命周期
功能特性
Pipeline提供了以下核心特性：
灵活的数据流定义：支持上层应用灵活定义数据处理流程
统一的processor接口：提供处理流程统一的processor结构体，便于统一管理
多种运行模式：支持独立任务、预设任务、中断处理、定时器处理等多种运行模式
数据流分支与合并：通过mirror和gather模块实现复杂的数据流拓扑
数据帧长度转换：通过shaper模块实现灵活的数据帧长度转换
动态参数更新：运行过程中可动态更新processor参数
流控与挂起：支持数据流的暂停和恢复
内存管理优化：提供scratch memory共享机制，减少内存碎片
架构设计
核心概念
Pipeline框架由以下核心概念组成：
1. Processor（处理器）
Processor是Pipeline中最基本的处理单元，代表一个独立的数据处理功能模块。每个processor需要实现标准的 wq_pipeline_processor_t 接口：
typedef struct pipeline_processor {
char *name; // Processor名称
WQ_RET (*query)(void *, struct pipeline_processor_info *); // 查询processor信息
void *(*probe)(void *, struct pipeline_processor_preset *); // 初始化processor
void (*remove)(void *); // 移除processor
void (*reset)(void *); // 重置processor
WQ_RET (*update)(void *, void *); // 更新processor参数
WQ_RET (*run)(void *, struct wq_plb *, struct wq_plb **); // 执行处理
} wq_pipeline_processor_t;
Processor接口说明：
query：查询processor的资源需求，返回所需MIPS、persistent buffer大小、scratch buffer大小等信息
probe：初始化processor，返回processor的handler（私有数据结构指针）
remove：清理processor资源
reset：重置processor到初始状态
update：动态更新processor的运行参数
run：实际的数据处理函数，输入一个plb，输出一个或多个plb
2. Procedure（处理工序）
Procedure是processor在特定stream中的实例化表示。当stream创建时，会为schedule中的每个processor创建对应的procedure。Procedure包含：
processor handler（由probe返回的processor私有数据）
processor配置信息
在stream中的索引位置
处理状态
Procedure是stream中实际执行的数据处理单元。
3. PLB（Pipeline Buffer）
PLB（Pipeline Buffer）是数据在Pipeline中流动的基本单位，包含数据帧的元数据和实际数据：
typedef struct wq_plb {
// 链表节点
struct wq_plb *next;
struct wq_plb *prev;
// 缓冲区指针
uint8_t *buf; // 数据缓冲区起始指针
uint8_t *end; // 数据缓冲区结束指针
uint8_t *data; // 实际数据起始指针
uint32_t sp; // 采样点数
uint32_t len; // 每个通道的数据长度（字节）
uint32_t channel_gap; // 通道间间隔（字节）
// 时间戳
uint32_t ts; // 主路径时间戳（采样点数）
uint32_t ts_sec; // 辅助路径时间戳
// 格式与通道
uint8_t format; // 数据格式（PCM/SBC/AAC等）
uint8_t channel; // 通道位图
uint8_t bit_width; // 采样位宽（16bit/32bit）
uint8_t dev_form; // 音频设备形式
uint8_t reserved;
// 用户计数
uint32_t users; // 引用计数
// 流程控制
void *procedure; // 挂起该plb的procedure
void (*free_cb)(struct wq_plb *, void *);
void *free_cb_param;
uint16_t plb_id; // plb关联的processor id和stream id
uint32_t param; // 用户定义参数
uint32_t flags; // 标志位
} wq_plb_t;
PLB标志位说明：
低16位（不继承）：
WQ_PIPELINE_BUFF_FLAG_ZERO_DATA：零数据标志WQ_PIPELINE_BUFF_FLAG_PB_DATA：PB数据标志WQ_PIPELINE_BUFF_FLAG_FORCE_STEREO_DATA：强制立体声数据标志
高16位（继承传递）：
WQ_PIPELINE_BUFF_FLAG_TAIL_DATA：尾部数据标志WQ_PIPELINE_BUFF_FLAG_MIX_DATA：混音数据标志WQ_PIPELINE_BUFF_FLAG_REF_DATA：参考数据标志WQ_PIPELINE_BUFF_FLAG_READ_ONLY：只读数据标志
4. Stream（数据流）
Stream是Pipeline中的数据流管理单元，包含：
一串按顺序执行的procedure链表
待处理的plb列表（plb_list）
流的运行状态和控制信息
可能拥有的独立任务或关联的预设任务
共享的scratch memory
Stream类型：
typedef enum {
WQ_PIPELINE_STREAM_CTX_OWN_TASK, // Stream拥有独立任务，可被触发
WQ_PIPELINE_STREAM_CTX_PRESET_TASK, // Stream属于预设任务，可被触发
WQ_PIPELINE_STREAM_CTX_IRQ_HANDLER, // Stream可被中断处理器触发
WQ_PIPELINE_STREAM_CTX_TIMER_HANDLER, // Stream可被定时器处理器触发
WQ_PIPELINE_STREAM_CTX_HIGH_SHARE_TASK, // Stream属于高优先级共享任务
WQ_PIPELINE_STREAM_CTX_LOW_SHARE_TASK, // Stream属于低优先级共享任务
WQ_PIPELINE_STREAM_CTX_CALLED, // Stream只能被调用，不能被触发
} WQ_PIPELINE_STREAM_CTX;
流程图
Pipeline的数据流处理流程如下：
flowchart TD
A([输入数据]) --> B[Procedure 1]
B --> buf1[/Pipeline buffer/]
buf1 --> C[Procedure 2]
C --> buf2[/Pipeline buffer/]
buf2 --> D[Procedure 3]
D --> buf3[/Pipeline buffer/]
buf3 --> E([输出数据])
style A fill:#e1f5e1,stroke:#4caf50,stroke-width:2px
style E fill:#ffe0e0,stroke:#f44336,stroke-width:2px
style B fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style C fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style D fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style buf1 fill:#fff3e0,stroke:#ff9800,stroke-width:2px
style buf2 fill:#fff3e0,stroke:#ff9800,stroke-width:2px
style buf3 fill:#fff3e0,stroke:#ff9800,stroke-width:2px
Pipeline数据流流程图
Pipeline使用时，每个处理步骤都作为独立的处理工序，以流水线方式运行。前一个处理工序的输出内容，作为下一个处理工序的输入内容。
运行原理
Stream创建流程
Stream创建命令接收：通过`wq_pipeline_stream_control()`函数发送CREATE命令
分配Stream资源： - 分配stream结构体 - 初始化procedure链表、command链表、plb列表 - 分配processor preset结构
创建Procedure： - 遍历stream_program数组 - 对每个processor调用probe接口，创建procedure - 将procedure加入stream的procedure链表
创建运行环境： - 根据stream类型创建独立任务或关联预设任务 - 创建stream互斥锁
分配Scratch Memory（可选）： - 如果启用了CONFIG_SCRATCH_MEM_ENABLE - 遍历所有processor，获取最大的scratch_size - 分配共享的scratch memory
Stream运行流程
运行触发
Stream的运行由以下方式触发：
显式触发：调用`wq_pipeline_stream_trigger(stream_id)`
任务循环：对于OWN_TASK类型的stream，任务持续运行
消息触发：通过IPC消息或中断触发
挂起数据检测：检测到有挂起的plb时自动触发
数据处理循环
Stream运行时的核心逻辑：
数据处理循环流程:
检查是否有挂起的数据帧 - 如果有挂起的plb，根据plb->procedure开始执行 - 如果没有挂起的plb，从stream的第一个procedure开始执行
按procedure链表顺序执行 - 遍历 stream->procedure_list - 调用 procedure->processor->run(procedure->handler, input_plb, &output_plb)
处理输出数据帧：如果procedure输出多个plb，第一个输出plb作为下一个procedure的输入，其他plb挂起到对应的procedure（plb->procedure赋值）
继续执行直到plb_list为空 - 重复步骤1-3，直到没有待处理的数据帧
Stream控制命令
Pipeline提供多种控制命令，通过`wq_pipeline_stream_control()`发送：
1. CREATE命令（创建流）
command.type = WQ_PIPELINE_STREAM_COMMAND_CREATE;
command.stream_id = STREAM_ID;
command.param = &runner;
command.init_sample_rate = sample_rate;
执行方式：同步阻塞执行，创建完成后才返回
功能：创建新的stream，初始化所有procedure
返回：创建成功后返回WQ_RET_OK
2. DESTROY命令（销毁流）
command.type = WQ_PIPELINE_STREAM_COMMAND_DESTROY;
command.stream_id = STREAM_ID;
command.callback = optional_destroy_callback;
command.callback_param = optional_event;
执行方式：默认异步执行
阻塞方式：实现callback函数，在callback中设置event，主线程等待event
功能：释放stream所有资源，包括procedure、plb、scratch memory
3. UPDATE_PROCEDURE命令（更新过程）
command.type = WQ_PIPELINE_STREAM_COMMAND_UPDATE_PROCEDURE;
command.stream_id = STREAM_ID;
command.processor = target_processor;
command.procedure_index = 0; // 保留字段
command.stream_depend = false; // 命令是否依赖stream运行
command.param = update_data;
command.callback = optional_callback;
command.callback_param = optional_param;
执行方式：异步执行
功能：动态更新指定processor的运行参数
参数传递：param传递给processor->update接口
4. 其他控制命令
PEND命令：暂停stream运行
RESUME命令：恢复stream运行
RESET_PROCEDURE命令：重置procedure状态
BYPASS_PROCEDURE命令：绕开procedure
DEBUG_PROCEDURE命令：启用procedure日志
DUMP_PROCEDURE_INPUT命令：启用procedure输入dump
DUMP_PROCEDURE_OUTPUT命令：启用procedure输出dump
数据流管理
PLB生命周期
PLB生命周期:
Processor在run()中调用wq_plb_alloc()分配plb
设置plb的数据指针、长度、格式等信息
将plb返回给stream作为输出
- PLB传递:
Stream将procedure输出的plb作为下一个procedure的输入
如果procedure输出多个plb： - 第一个plb直接传递给下一个procedure - 其他plb挂起（plb->procedure赋值）
- PLB释放:
Stream或processor调用wq_plb_free()释放plb
减少users引用计数
当users为0时，真正释放内存
如果设置了free_cb，调用free_cb回调
挂起数据帧
当processor需要等待外部事件时，可以将plb挂起：
// 将plb挂起到当前procedure
input_plb->procedure = current_procedure;
// 不将plb传递给下一个procedure，返回NULL
*output_plb = NULL;
当外部事件就绪后，再次触发stream，stream会根据plb->procedure从挂起点恢复执行。
Scratch Memory机制
Scratch memory是stream中所有processor共享的可复用内存区域：
优点： - 减少内存分配次数 - 降低内存碎片 - 提高内存使用效率
配置方式： 1. 自动分配：runner->scratch为NULL或scratch_size为0时，自动分配 2. 外部预分配：runner配置预分配的scratch memory
使用规则： - Scratch memory是stream中所有processor共享 - Processor必须在执行完成后不需要再使用scratch memory中的数据 - 不同processor按需使用scratch memory的不同区域 - Stream销毁时自动释放
多路数据流处理
Mirror模块
Mirror模块用于将一个procedure的输出分发到多个下游procedure：
应用场景： - 创建数据流旁路或回路 - 在不同stream之间传递数据 - 实现数据流广播
配置结构：
typedef struct {
uint16_t stream_id; // 当前所属stream的id
uint16_t mirror_id; // 预分配的id，相同mirror_id关联到同一mirror handler
uint16_t depth; // mirror tx_list的深度，超限会打印"process too slow"日志
bool copy_mode; // 是否复制数据，false时只clone plb配置
bool redirect_mode; // 是否丢弃数据
uint8_t bypass_id_map; // 需要丢弃数据的stream对应的map值
} wq_pipeline_mirror_config_t;
工作原理： 1. 每个mirror procedure都有一个mirror client 2. Mirror processor将每个mirror client的输入数据复制分发 3. 根据mirror_id将多个mirror client关联到同一mirror handler
Gather模块
Gather模块是多输入单输出的模块，用于混音运算：
应用场景： - 收集多个procedure的输出完成混音 - 多路音频数据合成
配置结构：
typedef struct {
uint16_t stream_id; // 当前所属stream的id
uint16_t bit_width; // 采样位宽（16bit/32bit）
uint16_t depth; // hold stb队列深度，0使用默认值
uint8_t gather_id; // 预分配id，相同gather_id的procedure关联在一起
bool primary; // 是否是输出流，只有primary gather才进行gather输出
bool match_ts; // 是否需要时间戳对齐
bool use_sec_ts; // 是否使用辅助路径时间戳
bool redirect_drop; // 非primary流rx_list满时是否丢弃数据
wq_pipeline_processor_t *sub_processor; // 子processor
void *sub_processor_config; // 子processor配置
} wq_pipeline_gather_config_t;
工作原理： 1. Gather client缓存输入plb到队列 2. Primary gather执行时，等待所有gather client的数据就绪 3. 根据match_ts进行时间戳对齐 4. 对齐的数据进行混音运算 5. 如果有sub_processor，传递给sub_processor处理
特殊行为： - 只有一个gather client时，gather processor不做处理，直接输出 - 非primary的gather client只缓存，不输出数据
Shaper模块
Shaper模块用于数据帧长度转换：
应用场景： - 改变数据帧的时长（如10ms → 7.5ms） - 统一不同来源的数据帧长度 - 调整算法处理粒度
配置结构：
typedef struct {
uint16_t size; // Pool大小（字节），0表示使用默认大小
uint16_t out_sp; // 输出采样点数（如16K采样率，7.5ms为120）
uint16_t bit_width; // 采样位宽（16bit/32bit）
uint16_t channel_num; // 通道数
} wq_pipeline_shaper_config_t;
工作原理： 1. 将输入的plb数据拷贝到预分配的数据池 2. 根据out_sp配置取出特定长度的数据帧 3. 输出数据帧，丢弃不完整的数据或缓存等待 4. 循环执行直到输入数据处理完
Breaker模块
Breaker模块用于释放上一个procedure输出的数据帧：
应用场景： - 在stream结束时释放不再需要的数据帧 - 避免内存泄漏 - 清理数据流末端
使用位置： - 必须放在stream_program END之前 - 作为stream的最后一个procedure
详细使用流程
步骤1：定义数据流处理流程
需要将音频处理流程分解为多个processor步骤，使用宏定义创建stream schedule。
简单流程（无分支）
适用于线性处理流程，每个procedure的输出只传递给下一个procedure：
flowchart TD
A([开始]) --> B[Processor A]
B --> C[Processor B]
C --> D[Processor C]
D --> E([结束])
style A fill:#e1f5e1,stroke:#4caf50,stroke-width:2px
style E fill:#ffe0e0,stroke:#f44336,stroke-width:2px
style B fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style C fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style D fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
线性数据流示意
#define STREAM_SCHEDULE_DEMO \
WQ_PIPELINE_STREAM_SCHEDULE_START \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_A, &A_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_B, &B_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_C, &C_config) \
WQ_PIPELINE_STREAM_SCHEDULE_END
void demo_stream_init(void)
{
wq_pipeline_stream_schedule_t demo_stream_schedule[] = STREAM_SCHEDULE_DEMO;
...
}
分支流程（使用Mirror）
适用于需要数据分流或回流的场景，使用mirror实现数据分发：
flowchart TD
A([开始]) --> B[Processor A]
B --> C[Processor B]
C --> D[Processor C]
D --> E[Processor D]
E --> M1{{Mirror 1}}
M1 --> F[Processor E]
F --> M2{{Mirror 2}}
M2 --> G[Processor F]
G --> H[Processor G]
H --> I[Processor H]
I --> J([结束])
M1 -.-> K[Processor I]
K -.-> B
M2 -.-> L[Processor J]
L -.-> H
style A fill:#e1f5e1,stroke:#4caf50,stroke-width:2px
style J fill:#ffe0e0,stroke:#f44336,stroke-width:2px
style M1 fill:#fff3e0,stroke:#ff9800,stroke-width:3px
style M2 fill:#fff3e0,stroke:#ff9800,stroke-width:3px
style B fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style C fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style D fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style E fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style F fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style G fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style H fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style I fill:#e3f2fd,stroke:#2196f3,stroke-width:2px
style K fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
style L fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
Mirror分支与回路示意
#define STREAM_SCHEDULE_DEMO \
WQ_PIPELINE_STREAM_SCHEDULE_START \
/* primary stream */ \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_A, &A_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_B, &B_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_C, &C_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_D, &D_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_1_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_E, &E_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_2_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_F, &F_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_G, &G_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_H, &H_config) \
/* sub stream 1 from mirror 1 */ \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_1_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_I, &I_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_B, &B_config) \
/* sub stream 2 from mirror 2 */ \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_2_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_J, &J_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_G, &G_config) \
WQ_PIPELINE_STREAM_SCHEDULE_END
void demo_stream_init(void)
{
wq_pipeline_stream_schedule_t demo_stream_schedule[] = STREAM_SCHEDULE_DEMO;
...
}
独立Sub Stream配置
也可以将sub stream配置在独立的stream中，提高可读性和维护性：
#define STREAM_SCHEDULE_DEMO \
WQ_PIPELINE_STREAM_SCHEDULE_START \
/* primary stream */ \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_A, &A_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_B, &B_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_C, &C_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_D, &D_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_1_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_E, &E_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_2_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_F, &F_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_G, &G_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_H, &H_config) \
WQ_PIPELINE_STREAM_SCHEDULE_END
#define STREAM_SCHEDULE_DEMO_SUB_STREAM_1 \
WQ_PIPELINE_STREAM_SCHEDULE_START \
/* sub stream1 from mirror 1 */ \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_1_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_I, &I_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_B, &B_config) \
WQ_PIPELINE_STREAM_SCHEDULE_END
#define STREAM_SCHEDULE_DEMO_SUB_STREAM_2 \
WQ_PIPELINE_STREAM_SCHEDULE_START \
/* sub stream2 from mirror 2 */ \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_mirror, &mirror_2_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_J, &J_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_G, &G_config) \
WQ_PIPELINE_STREAM_SCHEDULE_END
步骤2：实现自定义Processor
自定义processor需要实现`wq_pipeline_processor_t`接口：
/* Processor信息查询 */
static WQ_RET demo_processor_query(void *config, struct pipeline_processor_info *info)
{
demo_config_t *cfg = (demo_config_t *)config;
info->mips = 10; // processor需要的MIPS
info->persistent_size = 1024; // 持久化buffer大小
info->scratch_size = 512; // scratch buffer大小
info->stack_size = 256; // 栈大小
info->processor_info = NULL; // 自定义信息指针
return WQ_RET_OK;
}
/* Processor初始化 */
static void *demo_processor_probe(void *config, struct pipeline_processor_preset *preset)
{
demo_config_t *cfg = (demo_config_t *)config;
demo_handler_t *handler;
// 分配handler
handler = (demo_handler_t *)wq_heap_caps_malloc(
sizeof(demo_handler_t), MALLOC_CAP_NONE);
assert(handler);
// 初始化handler
memset(handler, 0, sizeof(demo_handler_t));
handler->config = cfg;
handler->preset = preset;
// 初始化内部资源
/* ... */
return handler;
}
/* Processor移除 */
static void demo_processor_remove(void *handler)
{
demo_handler_t *h = (demo_handler_t *)handler;
// 释放内部资源
/* ... */
// 释放handler
wq_heap_caps_free(h);
}
/* Processor重置 */
static void demo_processor_reset(void *handler)
{
demo_handler_t *h = (demo_handler_t *)handler;
// 重置内部状态
/* ... */
}
/* Processor参数更新 */
static WQ_RET demo_processor_update(void *handler, void *param)
{
demo_handler_t *h = (demo_handler_t *)handler;
demo_update_t *update = (demo_update_t *)param;
// 更新processor参数
h->config->param = update->param;
return WQ_RET_OK;
}
/* Processor执行 */
static WQ_RET demo_processor_run(void *handler, struct wq_plb *input_plb,
struct wq_plb **output_plb)
{
demo_handler_t *h = (demo_handler_t *)handler;
uint32_t out_len = calculate_output_len(input_plb);
// 分配输出plb
*output_plb = wq_plb_alloc(out_len, 0);
if (!*output_plb) {
return WQ_RET_NOMEM;
}
// 设置输出plb属性
(*output_plb)->format = input_plb->format;
(*output_plb)->channel = input_plb->channel;
(*output_plb)->bit_width = input_plb->bit_width;
(*output_plb)->sample_rate = input_plb->sample_rate;
// 处理数据
memcpy((*output_plb)->data, input_plb->data, input_plb->len);
/* 数据处理逻辑... */
return WQ_RET_OK;
}
/* Processor接口定义 */
const wq_pipeline_processor_t demo_processor = {
.name = "demo_processor",
.query = demo_processor_query,
.probe = demo_processor_probe,
.remove = demo_processor_remove,
.reset = demo_processor_reset,
.update = demo_processor_update,
.run = demo_processor_run,
};
注意事项：
run()函数是核心，实现实际的数据处理逻辑
输出plb管理：run()必须分配输出plb，调用方负责释放
引用计数：通过wq_plb_get()和wq_plb_free()管理plb生命周期
错误处理：遇到错误时返回适当的WQ_RET错误码
线程安全：如果processor在多线程环境使用，需要适当的同步
步骤3：配置Stream Runner
配置stream的运行环境和任务参数：
/* Stream任务配置 */
static char demo_task_name[] = "stream_demo";
static wq_pipeline_stream_task_cfg_t demo_task_cfg = {
.prio = 7, // 任务优先级
.stack_size = 2048, // 任务栈大小
.name = demo_task_name, // 任务名称
};
void demo_stream_start(void)
{
/* 定义stream runner */
wq_pipeline_stream_runner_t demo_runner;
memset(&demo_runner, 0, sizeof(wq_pipeline_stream_runner_t));
demo_runner.type = WQ_PIPELINE_STREAM_CTX_OWN_TASK;
demo_runner.config = &demo_task_cfg;
demo_runner.schedule = demo_stream_schedule;
/* scratch memory配置可选，见下节 */
}
Stream类型选择：
WQ_PIPELINE_STREAM_CTX_OWN_TASK：推荐，stream拥有独立任务，可随时触发
WQ_PIPELINE_STREAM_CTX_PRESET_TASK：多个stream共享同一任务
WQ_PIPELINE_STREAM_CTX_CALLED：stream只能被函数调用，不能独立运行
步骤4：配置Scratch Memory
Scratch memory是stream中所有processor共享的可复用内存：
方式1：自动分配（推荐）
系统根据所有processor的最大scratch_size自动分配：
/* 开启CONFIG_SCRATCH_MEM_ENABLE后自动分配 */
#define CONFIG_SCRATCH_MEM_ENABLE
void demo_stream_start(void)
{
wq_pipeline_stream_runner_t demo_runner;
memset(&demo_runner, 0, sizeof(wq_pipeline_stream_runner_t));
demo_runner.type = WQ_PIPELINE_STREAM_CTX_OWN_TASK;
demo_runner.config = &demo_task_cfg;
demo_runner.schedule = demo_stream_schedule;
/* 不设置scratch和scratch_size，系统自动分配 */
}
优点： - 简单易用，无需手动管理 - 系统自动计算所需大小 - Stream销毁时自动释放
前提条件： - 需要定义CONFIG_SCRATCH_MEM_ENABLE宏 - Processor的query()必须正确返回scratch_size
方式2：外部预分配
对于有特殊内存要求的场景，可以外部预分配scratch memory：
void demo_stream_start(void)
{
/* 定义stream runner */
wq_pipeline_stream_runner_t demo_runner;
memset(&demo_runner, 0, sizeof(wq_pipeline_stream_runner_t));
demo_runner.type = WQ_PIPELINE_STREAM_CTX_OWN_TASK;
demo_runner.config = &demo_task_cfg;
demo_runner.schedule = demo_stream_schedule;
/* 外部预分配scratch memory */
void *demo_scratch;
demo_scratch = wq_heap_caps_malloc(sizeof(uint8_t), MALLOC_CAP_NONE);
assert(demo_scratch);
memset(demo_scratch, 0, sizeof(uint8_t));
demo_runner.scratch = demo_scratch;
demo_runner.scratch_size = sizeof(uint8_t);
/* 创建stream */
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
command.type = WQ_PIPELINE_STREAM_COMMAND_CREATE;
command.stream_id = STREAM_ID_DEMO;
command.init_sample_rate = demo_sample_rate;
command.param = &demo_runner;
wq_pipeline_stream_control(&command);
}
注意事项：
大小匹配：外部分配的scratch_size必须大于等于所有processor的scratch_size总和
内存属性：根据需要选择合适的内存属性（对齐、DMA等）
生命周期：外部scratch memory需要在stream销毁后手动释放
共享规则：scratch memory是stream中所有processor共享，需要保证使用安全
步骤5：创建Stream
发送CREATE命令创建stream：
void demo_stream_start(void)
{
wq_pipeline_stream_runner_t demo_runner;
memset(&demo_runner, 0, sizeof(wq_pipeline_stream_runner_t));
demo_runner.type = WQ_PIPELINE_STREAM_CTX_OWN_TASK;
demo_runner.config = &demo_task_cfg;
demo_runner.schedule = demo_stream_schedule;
/* 创建stream命令 */
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
command.type = WQ_PIPELINE_STREAM_COMMAND_CREATE;
command.stream_id = STREAM_ID_DEMO;
command.init_sample_rate = demo_sample_rate;
command.param = &demo_runner;
/* 发送CREATE命令（阻塞执行） */
WQ_RET ret = wq_pipeline_stream_control(&command);
if (ret != WQ_RET_OK) {
printf("Create stream failed: %d\n", ret);
return;
}
printf("Stream %d created successfully\n", STREAM_ID_DEMO);
}
CREATE命令执行流程：
阻塞执行，创建完成后才返回
分配stream结构体和内部资源
遍历schedule数组，为每个processor创建procedure
根据stream类型创建或关联任务
分配scratch memory（如果配置了）
初始化stream状态为有效
步骤6：启动Stream运行
Stream创建后，需要触发才能开始处理数据：
void demo_stream_start(void)
{
/* ... 创建stream ... */
/* 触发stream运行 */
WQ_RET ret = wq_pipeline_stream_trigger(STREAM_ID_DEMO);
if (ret != WQ_RET_OK) {
printf("Trigger stream failed: %d\n", ret);
return;
}
printf("Stream %d triggered\n", STREAM_ID_DEMO);
}
触发方式：
单次触发：调用`wq_pipeline_stream_trigger()`，stream处理完所有挂起数据后停止
持续触发：在数据到达时持续触发，stream保持活跃状态
定时触发：通过定时器定期触发，适用于需要定期处理的数据
OWN_TASK类型stream： - 拥有独立任务的stream会自动循环运行 - 不需要显式触发，但触发可加速处理
步骤7：动态更新Processor参数
运行过程中，可以通过UPDATE命令动态更新processor参数：
/* 更新命令回调，用于释放动态分配的参数 */
static void demo_update_callback(void *arg)
{
uint32_t *demo_update_param = (uint32_t *)arg;
wq_heap_caps_free(demo_update_param);
}
void demo_stream_update(void)
{
/* 准备更新参数 */
uint32_t *demo_update_param;
demo_update_param = wq_heap_caps_malloc(sizeof(uint8_t), MALLOC_CAP_NONE);
assert(demo_update_param);
memset(demo_update_param, 0, sizeof(uint8_t));
*demo_update_param = NEW_VALUE; // 设置新值
/* 配置更新命令 */
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
command.type = WQ_PIPELINE_STREAM_COMMAND_UPDATE_PROCEDURE;
command.stream_id = STREAM_ID_DEMO;
command.processor = &demo_processor; // 指定要更新的processor
command.procedure_index = 0; // 保留字段，可设为0
command.stream_depend = false; // 不依赖stream运行状态
command.param = demo_update_param; // 更新参数
command.callback = demo_update_callback; // 命令完成回调
command.callback_param = demo_update_param; // 回调参数（用于释放）
/* 发送UPDATE命令（异步执行） */
WQ_RET ret = wq_pipeline_stream_control(&command);
if (ret != WQ_RET_OK) {
printf("Update processor failed: %d\n", ret);
wq_heap_caps_free(demo_update_param);
}
}
UPDATE命令特点：
异步执行：命令加入stream的command队列，在stream运行时处理
参数传递：param传递给processor->update()接口
回调机制：完成后调用callback，可用于释放动态分配的参数
stream_depend：决定是否依赖stream运行，false时即使stream暂停也会处理
步骤8：销毁Stream
不再需要stream时，需要发送DESTROY命令：
异步销毁（默认）
void demo_stream_stop(void)
{
/* 销毁stream命令 */
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
command.type = WQ_PIPELINE_STREAM_COMMAND_DESTROY;
command.stream_id = STREAM_ID_DEMO;
/* 发送DESTROY命令（异步执行，立即返回） */
wq_pipeline_stream_control(&command);
printf("Stream %d destroy command sent\n", STREAM_ID_DEMO);
}
同步销毁（带回调）
如果需要等待stream销毁完成，需要实现回调函数：
/* stream销毁完成的回调 */
static void demo_stream_destroy_callback(void *arg)
{
os_event_h p_event = (os_event_h)arg;
os_set_event(p_event, BIT(0));
}
void demo_stream_stop(void)
{
/* 创建事件用于同步 */
os_event_h p_event;
p_event = os_create_event(WQ_PIPELINE_MID);
if (!p_event) {
assert(0);
}
/* 销毁stream命令 */
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
command.type = WQ_PIPELINE_STREAM_COMMAND_DESTROY;
command.stream_id = STREAM_ID_DEMO;
command.callback = demo_stream_destroy_callback;
command.callback_param = p_event;
/* 发送DESTROY命令 */
wq_pipeline_stream_control(&command);
/* 等待销毁完成事件 */
uint32_t events;
bool ret = os_wait_event(p_event, MAX_TIME, &events);
if (ret && (events & BIT(0))) {
printf("Stream %d destroyed successfully\n", STREAM_ID_DEMO);
}
os_delete_event(p_event);
}
同步销毁流程：
发送DESTROY命令
命令加入command队列，异步执行
主线程等待event
Stream销毁完成后，回调设置event
主线程从等待中返回
特殊模块使用详解
Pipeline Shaper使用
Shaper模块用于实现PCM数据帧长度转换：
配置示例（将10ms数据帧转换为7.5ms）：
/* 16K采样率 */
#define SAMPLE_RATE_16K 16000
/* 10ms = 160采样点 */
#define FRAME_SIZE_10MS 160
/* 7.5ms = 120采样点 */
#define FRAME_SIZE_7_5MS 120
wq_pipeline_shaper_config_t demo_shaper_config = {
.size = 0, // 使用默认缓存buffer长度
.out_sp = FRAME_SIZE_7_5MS, // 输出120个采样点（7.5ms）
.bit_width = 16, // 16bit位宽
.channel_num = 2, // 2通道
};
/* 在stream schedule中使用 */
WQ_PIPELINE_STREAM_SCHEDULE_STEP(wq_pipeline_shaper, &demo_shaper_config)
Shaper工作原理：
接收输入plb，拷贝数据到内部缓冲区（数据池）
按照out_sp配置的采样点数取出数据
如果缓冲区数据不足，缓存数据等待更多输入
输出固定长度的数据帧，保持输出帧长度的稳定性
注意事项：
size参数：0表示自动计算缓存大小，通常为4 * out_sp * bit_width/8 * channel_num
累积效应：输入数据会累积直到满足输出长度，可能引入延迟
延迟控制：根据应用场景选择合适的out_sp，平衡延迟和稳定性
Pipeline Gather使用
Gather模块用于多路数据混音：
配置示例：
wq_pipeline_gather_config_t demo_gather_config = {
.stream_id = STREAM_ID_DEMO,
.gather_id = STREAM_ID_GATHER,
.bit_width = 16,
.depth = 0, // 使用默认深度
.primary = true, // 这是主流，即输出流
.match_ts = true, // 需要时间戳对齐
.use_sec_ts = true, // 使用sec时间戳
.sub_processor = NULL, // 没有子处理流程
.sub_processor_config = NULL,
};
/* gather client配置（非primary） */
wq_pipeline_gather_config_t demo_gather_client_config = {
.stream_id = STREAM_ID_INPUT_1,
.gather_id = STREAM_ID_GATHER,
.bit_width = 16,
.depth = 0,
.primary = false, // 这是输入流
.match_ts = true,
.use_sec_ts = true,
.redirect_drop = true, // 队列满时丢弃数据
.sub_processor = NULL,
.sub_processor_config = NULL,
};
Gather工作原理：
Gather client接收输入： - Primary client：输入plb，等待其他client数据 - 非primary client：输入plb缓存到队列，输出NULL
时间戳对齐： - 根据match_ts配置，检查各输入的时间戳 - 如果use_sec_ts为true，使用ts_sec进行对齐 - 等待所有client数据对齐
混音处理： - Primary client触发时执行混音 - 将各client数据按比例相加 - 考虑位宽和通道数
输出数据： - 只有primary gather输出混音结果 - 可选传递给sub_processor进一步处理
注意事项：
gather_id必须一致：所有相关的gather使用相同的gather_id
必须有一个primary：只有primary gather会输出数据
redirect_drop配置：非primary流建议设置redirect_drop为true
depth配置：影响队列深度，影响对齐等待时间和内存占用
Pipeline Mirror使用
Mirror模块用于数据分发：
配置示例：
wq_pipeline_mirror_config_t demo_mirror_config = {
.stream_id = STREAM_ID_DEMO,
.mirror_id = STREAM_ID_MIRROR,
.depth = 0, // 使用默认深度4
.copy_mode = false, // 只clone plb配置，不复制数据
.redirect_mode = true, // 启用丢弃模式
.bypass_id_map = BIT(STREAM_ID_BYPASS), // 旁路stream需要丢弃数据
};
Mirror工作原理：
每个mirror procedure是一个mirror client
Mirror processor维护tx_list，存储每个client的待发送数据
当有client数据输入时： - 如果copy_mode为false，只clone plb配置 - 如果copy_mode为true，复制整个plb数据
根据mirror_id，数据分发到所有关联的client
检查tx_list深度，超过depth时打印"process too slow"
性能考虑：
copy_mode选择： - false：高效，只复制plb头，适用于只读场景 - true：完整复制，适用于需要修改plb的场景
depth配置：影响tx_list最大深度，默认为4 - 深度太小：容易超限，影响处理速度 - 深度太大：占用更多内存
redirect_mode：用于数据流控制，可动态丢弃某些stream的数据
常见问题及解决方案
问题1：Stream创建失败
现象：wq_pipeline_stream_control`返回`WQ_RET_INVAL`或`WQ_RET_NOMEM
可能原因： 1. stream_id超出范围（>= CONFIG_PIPELINE_STREAM_MAX_NUM） 2. runner->config为NULL 3. stream已存在（重复创建） 4. 内存不足（procedure创建失败） 5. scratch memory分配失败
解决方案：
检查stream_id范围
#define STREAM_ID_DEMO 0 // 确保小于CONFIG_PIPELINE_STREAM_MAX_NUM
检查runner配置
demo_runner.config = &demo_task_cfg; // 必须非NULL
检查stream是否已存在
// 创建前检查
if (wq_pipeline_stream_is_created(STREAM_ID_DEMO)) {
printf("Stream already exists\n");
return;
}
增加系统内存：检查heap大小，调整内存配置 减小scratch_size需求：优化processor的scratch使用
问题2：Stream运行卡住
现象： - 调用`wq_pipeline_stream_trigger()`后stream没有响应 - Stream处于挂起状态，不处理数据
可能原因： 1. Stream未创建成功 2. Stream中没有待处理的plb 3. Procedure陷入死循环 4. Scratch memory损坏导致processor异常
解决方案：
验证stream创建状态
WQ_RET ret = wq_pipeline_stream_trigger(STREAM_ID_DEMO);
if (ret == WQ_RET_NOT_EXIST) {
printf("Stream not exist\n");
return;
}
检查plb输入
// 确保有plb输入到stream
wq_plb_t *input_plb = wq_plb_alloc(len, 0);
if (!input_plb) {
return;
}
// 将plb添加到stream的plb_list
// 可以通过特定的processor实现plb注入
启用调试日志
// 发送DEBUG_PROCEDURE命令
command.type = WQ_PIPELINE_STREAM_COMMAND_DEBUG_PROCEDURE;
command.processor = &problematic_processor;
wq_pipeline_stream_control(&command);
检查processor实现 - 确保run()函数有返回路径 - 避免无限循环 - 检查条件判断是否正确
问题3：数据帧丢失
现象： - 某些processor的输入数据帧丢失 - 输出数据不连续
可能原因： 1. PLB引用计数错误 2. Processor输出多个plb时处理不当 3. Stream挂起的plb没有正确恢复 4. Mirror或Gather配置错误
解决方案：
正确管理PLB引用
static WQ_RET demo_processor_run(void *handler, struct wq_plb *input_plb,
struct wq_plb **output_plb)
{
// 增加引用计数
wq_plb_get(input_plb);
// 处理数据
*output_plb = wq_plb_alloc(out_len, 0);
// 传递引用给输出
(*output_plb)->users = input_plb->users;
return WQ_RET_OK;
}
正确处理多输出场景
static WQ_RET multi_output_processor_run(void *handler, struct wq_plb *input_plb,
struct wq_plb **output_plb)
{
// 假设输出3个plb
wq_plb_t *out1, *out2, *out3;
// 分配输出plb
out1 = wq_plb_alloc(len1, 0);
out2 = wq_plb_alloc(len2, 0);
out3 = wq_plb_alloc(len3, 0);
// 设置第一个plb作为直接输出
*output_plb = out1;
// 挂起其他plb（stream会管理）
out2->procedure = next_procedure;
out3->procedure = next_procedure;
return WQ_RET_OK;
}
检查Mirror/Gather配置 - 确保mirror_id和gather_id一致性 - 检查primary设置是否正确 - 验证depth配置是否足够
问题4：内存泄漏
现象： - 系统运行一段时间后内存持续增长 - Heap内存不足
可能原因： 1. PLB未正确释放 2. Procedure handler未正确清理 3. Scratch memory未释放 4. UPDATE命令的动态参数未释放
解决方案：
确保PLB释放
// 在processor中确保释放输入plb
static WQ_RET demo_processor_run(void *handler, struct wq_plb *input_plb,
struct wq_plb **output_plb)
{
*output_plb = wq_plb_alloc(out_len, 0);
// 减少输入plb的引用
wq_plb_free(input_plb); // 引用计数减1
return WQ_RET_OK;
}
实现remove接口
static void demo_processor_remove(void *handler)
{
demo_handler_t *h = (demo_handler_t *)handler;
// 释放handler内部分配的资源
if (h->internal_buf) {
wq_heap_caps_free(h->internal_buf);
}
// 释放handler
wq_heap_caps_free(h);
}
使用callback释放UPDATE参数
static void demo_update_callback(void *arg)
{
// 释放UPDATE命令的参数
uint32_t *param = (uint32_t *)arg;
wq_heap_caps_free(param);
}
检查Stream销毁
// 确保调用DESTORY命令
command.type = WQ_PIPELINE_STREAM_COMMAND_DESTROY;
command.stream_id = STREAM_ID_DEMO;
command.callback = destroy_callback;
wq_pipeline_stream_control(&command);
问题5：性能问题
现象： - "process too slow"日志频繁打印 - 音频卡顿或延迟过大 - CPU使用率过高
可能原因： 1. Processor执行时间过长 2. PLB拷贝开销过大 3. Mirror/Gather深度配置不合理 4. 任务优先级冲突
解决方案：
优化Processor性能
// 使用DMA减少CPU拷贝
// 优化算法复杂度
// 使用查表替代计算
// 减少内存分配次数
启用copy_mode优化
wq_pipeline_mirror_config_t mirror_config = {
.copy_mode = false, // 只clone配置，不复制数据
// ...
};
调整Mirror/Gather深度
wq_pipeline_mirror_config_t mirror_config = {
.depth = 8, // 增加深度，减少"process too slow"
// ...
};
合理设置任务优先级
wq_pipeline_stream_task_cfg_t task_cfg = {
.prio = 7, // 音频通常为中等优先级
// ...
};
使用Scratch Memory
// 启用scratch memory减少分配
#define CONFIG_SCRATCH_MEM_ENABLE
问题6：时间戳不对齐
现象： - Gather混音效果异常 - 音频卡顿或延迟不均匀
可能原因： 1. match_ts配置错误 2. use_sec_ts配置与实际情况不符 3. 输入数据时间戳不连续
解决方案：
检查Gather配置
wq_pipeline_gather_config_t gather_config = {
.match_ts = true, // 确保启用时间戳对齐
.use_sec_ts = false, // 根据实际情况选择主或辅时间戳
// ...
};
验证输入时间戳
// 检查输入plb的时间戳
printf("ts: %u, ts_sec: %u\n", input_plb->ts, input_plb->ts_sec);
检查Gather client数量 - 确保所有相关client都加入了gather - 检查是否有client丢失数据
问题7：多路数据流冲突
现象： - 多个stream同时运行时数据混乱 - 不同stream的数据互相干扰
可能原因： 1. Stream ID配置冲突 2. Mirror ID或Gather ID冲突 3. Procedure共享了不当的handler
解决方案：
确保Stream ID唯一
#define STREAM_ID_DEMO_1 0
#define STREAM_ID_DEMO_2 1
// 确保每个stream使用不同的ID
区分Mirror/Gather ID
// 不同stream使用不同的mirror_id和gather_id
wq_pipeline_mirror_config_t mirror1_config = {
.mirror_id = 0,
// ...
};
wq_pipeline_mirror_config_t mirror2_config = {
.mirror_id = 1, // 不同的mirror_id
// ...
};
检查Processor实现 - 确保每个stream的probe创建独立的handler - 避免不同stream共享同一handler
问题8：PLB格式错误
现象： - Processor无法正确解析数据 - 输出音频异常
可能原因： 1. PLB format、channel、bit_width设置错误 2. 数据长度计算错误 3. 通道配置与实际数据不匹配
解决方案：
正确设置PLB属性
static WQ_RET demo_processor_run(void *handler, struct wq_plb *input_plb,
struct wq_plb **output_plb)
{
// 继承输入plb的主要属性
*output_plb = wq_plb_alloc(out_len, 0);
(*output_plb)->format = input_plb->format;
(*output_plb)->channel = input_plb->channel;
(*output_plb)->bit_width = input_plb->bit_width;
(*output_plb)->sample_rate = input_plb->sample_rate;
// 设置数据长度
(*output_plb)->len = input_plb->len;
return WQ_RET_OK;
}
计算正确的数据长度
// 计算输出长度（字节）
uint32_t out_len_bytes = out_samples * bit_width / 8 * channels;
// 设置plb长度
output_plb->len = out_len_bytes;
验证通道配置
// 检查输入plb的通道位图
if (input_plb->channel & WQ_PIPELINE_CHANNEL_LEFT) {
// 处理左通道
}
if (input_plb->channel & WQ_PIPELINE_CHANNEL_RIGHT) {
// 处理右通道
}
问题9：Pipeline shaper异常
现象： - shaper_run里wq_plb_get_channels(in) != shaper->channel_num触发assert
可能原因： 1. shaper配置的channel num和实际输入plb的channel num不一致
解决方案：
确保shaper配置的channel num和数据流在运行过程中输入plb的channel num一致
问题10：Pipeline gather异常
现象： - gather_stereo_run里temp != NULL && in->len != temp->len触发assert
可能原因： 1. 需要gather到一起的两个数据长度不一致
解决方案：
确保需要gather到一起的两个数据长度一致,可在某个gather_processor前增加shaper_processor,将shaper_processor的输出长度设置为和另一个数据流长度一致
#define STREAM_SCHEDULE_DEMO_STREAM_1 \
WQ_PIPELINE_STREAM_SCHEDULE_START \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_A, &A_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(shaper_processor, &shaper_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(gather_processor, &gather_config) \
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor_B, &B_config) \
WQ_PIPELINE_STREAM_SCHEDULE_END
调试技巧
启用Processor调试信息
Pipeline支持对单个procedure启用调试信息：
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
// 启用procedure调试日志
command.type = WQ_PIPELINE_STREAM_COMMAND_DEBUG_PROCEDURE;
command.stream_id = STREAM_ID_DEMO;
command.processor = &demo_processor;
wq_pipeline_stream_control(&command);
启用后会打印procedure的输入输出信息，包括plb地址、长度、格式等。
Dump PLB数据
可以dump procedure的输入或输出plb数据：
// Dump procedure输入plb
command.type = WQ_PIPELINE_STREAM_COMMAND_DUMP_PROCEDURE_INPUT;
command.stream_id = STREAM_ID_DEMO;
command.processor = &demo_processor;
wq_pipeline_stream_control(&command);
// Dump procedure输出plb
command.type = WQ_PIPELINE_STREAM_COMMAND_DUMP_PROCEDURE_OUTPUT;
command.stream_id = STREAM_ID_DEMO;
command.processor = &demo_processor;
wq_pipeline_stream_control(&command);
运行时状态查询
可以查询stream的运行状态：
wq_pipeline_stream_command_t command;
memset(&command, 0, sizeof(wq_pipeline_stream_command_t));
command.type = WQ_PIPELINE_STREAM_COMMAND_QUERY;
command.stream_id = STREAM_ID_DEMO;
wq_pipeline_stream_control(&command);
自定义PLB释放回调
可以为plb设置释放回调，跟踪PLB生命周期：
// PLB释放回调
void my_plb_free_callback(struct wq_plb *plb, void *param)
{
printf("PLB %p freed, size: %u\n", plb, plb->len);
}
// 使用PLB时设置回调
wq_plb_t *plb = wq_plb_alloc(size, 0);
plb->free_cb = my_plb_free_callback;
plb->free_cb_param = NULL;
性能优化建议
减少PLB拷贝 - 尽可能使用PLB直接传递，避免不必要的数据复制 - 使用copy_mode=false的mirror配置 - 考虑使用PLB共享机制
合理使用Scratch Memory - 启用CONFIG_SCRATCH_MEM_ENABLE - 将频繁分配的临时数据放到scratch memory - 避免在run()中频繁分配内存
优化Processor算法 - 使用查表替代复杂计算 - 使用SIMD指令加速（如果平台支持） - 减少循环次数
合理配置任务优先级 - 音频stream通常使用中高优先级（5-8） - 避免与实时任务冲突 - 考虑使用预设任务减少任务切换开销
控制数据帧大小 - 选择适当的数据帧长度平衡延迟和处理效率 - 过大的帧增加延迟，过小增加调度开销 - 典型值：7.5ms、10ms、15ms
批量处理 - 在条件允许时，批量处理多个plb - 减少循环次数 - 提高cache命中率
最佳实践
Processor设计 - 保持职责单一，每个processor只做一件事 - 使用标准的query/probe/remove/update/run接口 - 正确返回错误码，避免assert - 充分利用scratch memory
Stream设计 - 将相关processor组织到同一stream - 合理安排processor顺序，减少数据转换 - 使用mirror和gather实现复杂拓扑 - 为stream设置合理的优先级和栈大小
错误处理 - 所有接口返回正确的WQ_RET错误码 - 失败时回滚已分配的资源 - 在callback中清理动态分配的参数 - 记录关键错误日志
资源管理 - 使用PLB引用计数管理生命周期 - 及时释放不再使用的PLB - 避免内存泄漏 - 合理使用scratch memory减少碎片
文档和注释 - 为processor添加清晰的doxygen注释 - 说明processor的功能、输入输出ㇺ、配置参数 - 记录重要的设计决策 - 提供使用示例
参考示例
examples/pipeline_demo
更多示例代码可参考pipeline_demo项目，其中包含： - SBC/MSBC解码流示例 - Reader/Writer processor实现 - Stream创建和销毁示例 - 动态更新示例
API 介绍
Defines
-
WQ_PIPELINE_BUFF_DEFAULT_PAD_SIZE
Pad size
-
WQ_PIPELINE_BUFF_FLAG_ZERO_DATA
Zero data flag of pipeline buff
-
WQ_PIPELINE_BUFF_FLAG_PB_DATA
PB data flag of pipeline buff
-
WQ_PIPELINE_BUFF_FLAG_FORCE_STEREO_DATA
Force stereo data flag of pipeline buff
-
WQ_PIPELINE_BUFF_FLAG_TAIL_DATA
Tail data flag of pipeline buff
-
WQ_PIPELINE_BUFF_FLAG_MIX_DATA
Mix data flag of pipeline buff
-
WQ_PIPELINE_BUFF_FLAG_REF_DATA
Mic ref valid flag of pipeline buff
-
WQ_PIPELINE_BUFF_FLAG_READ_ONLY
Read only data flag of pipeline buff
-
WQ_PIPELINE_DATA_FORMAT_INVALID
Invalid format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_PCM
PCM format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_SBC
SBC format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_AAC
AAC format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_LDAC
LDAC format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_LHDC
LHDC format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_LC3
LC3 format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_MSBC
MSBC format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_CVSD
CVSD format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_OPUS
OPUS format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_COMPLEX
COMPLEX format of pipeline buff data
-
WQ_PIPELINE_DATA_FORMAT_L2HC
L2HC format of pipeline buff data
-
WQ_PIPELINE_CHANNEL_NUM_MAX
Max channel num
-
WQ_PIPELINE_CHANNEL_INVALID
Invalid channel num
-
WQ_PIPELINE_CHANNEL_DECODE_MASK
Invalid channel num
-
WQ_PIPELINE_CHANNEL_LEFT
Left channel flag
-
WQ_PIPELINE_CHANNEL_RIGHT
Right channel flag
-
WQ_PIPELINE_CHANNEL_MIX
Mix channel flag
-
WQ_PIPELINE_CHANNEL_DUAL
Dual channel flag
-
WQ_PIPELINE_CHANNEL_SINGLE
Single channel flag
-
WQ_PIPELINE_CHANNEL_MIC0
Mic0 channel flag
-
WQ_PIPELINE_CHANNEL_MIC1
Mic1 channel flag
-
WQ_PIPELINE_CHANNEL_MIC2
Mic2 channel flag
-
WQ_PIPELINE_CHANNEL_MIC3
Mic3 channel flag
-
WQ_PIPELINE_CHANNEL_MIC4
Mic4 channel flag
-
WQ_PIPELINE_CHANNEL_MIC5
Mic5 channel flag
-
WQ_PIPELINE_CHANNEL_MIC6
Mic6 channel flag
-
WQ_PIPELINE_CHANNEL_MIC7
Mic7 channel flag
-
WQ_PIPELINE_CHANNEL_STEREO
Stereo channel flag
-
WQ_PIPELINE_CHANNEL_LEFTDUAL
Left dual channel flag
-
WQ_PIPELINE_CHANNEL_RIGHTDUAL
Right dual channel flag
-
WQ_PIPELINE_CHANNEL_MIXDUAL
Mix dual channel flag
-
TWS_LEFT_DEV_TYPE
Tws left dev form
-
TWS_RIGHT_DEV_TYPE
Tws right dev form
-
AUD_STEREO_DEV_TYPE
Audio stereo dev form
-
AUD_MIXER_DEV_TYPE
Audio mixer dev form
-
WQ_PIPELINE_SAMPLE_WIDTH_16BIT
16 bit width of pipeline buff data
-
WQ_PIPELINE_SAMPLE_WIDTH_32BIT
32 bit width of pipeline buff data
Typedefs
Functions
-
static inline void wq_plb_list_init(wq_plb_list_t *list)
Init pipeline buff list head.
- 参数:
list -- The pipeline buff list head
-
static inline bool wq_plb_list_is_empty(const wq_plb_list_t *list)
Check if a queue is empty.
- 参数:
list -- The pipeline buff list head
- 返回:
True if the list is empty
-
static inline uint32_t wq_plb_list_get_len(const wq_plb_list_t *list)
Get pipeline buff list length.
- 参数:
list -- The pipeline buff list head
- 返回:
Queue len
-
static inline bool wq_plb_is_list_first(const wq_plb_list_t *list, const wq_plb_t *adb)
Check if an pipeline buff is the first entry in the list.
- 参数:
list -- The pipeline buff list head
adb -- Pipeline buff is current packet
- 返回:
True if the pipeline buff is the first entry
-
static inline bool wq_plb_is_list_last(const wq_plb_list_t *list, const wq_plb_t *adb)
Check if an pipeline buff is the last entry in the list.
- 参数:
list -- The pipeline buff list head
adb -- Pipeline buff is current packet
- 返回:
True if the pipeline buff is the last entry
-
static inline wq_plb_t *wq_plb_peek_head(const wq_plb_list_t *list)
Find the first pipeline buff in the list, without dequeue.
- 参数:
list -- The pipeline buff list head
- 返回:
First pipeline buff in the list
-
static inline wq_plb_t *wq_plb_peek_next(wq_plb_t *adb, const wq_plb_list_t *list)
Find the next pipeline buff in the list, without dequeue.
- 参数:
adb -- Pipeline buff is the current packet
list -- The pipeline buff list head
- 返回:
Next pipeline buff in the list
-
static inline wq_plb_t *wq_plb_peek_tail(const wq_plb_list_t *list)
Find the last pipeline buff in the list, without dequeue.
- 参数:
list -- The pipeline buff list head
- 返回:
Last pipeline buff in the list
-
static inline void wq_plb_insert(wq_plb_t *newst, wq_plb_t *prev, wq_plb_t *next, wq_plb_list_t *list)
Insert new pipeline buff between two old pipeline buffs.
备注
This function must be called atomically
- 参数:
newst -- New pipeline buff to be inserted
prev -- Prev pipeline buff
next -- Next pipeline buff
list -- The pipeline buff list head
-
static inline void wq_plb_queue_head(wq_plb_list_t *list, wq_plb_t *newst)
Queue new packet in the head of list.
- 参数:
list -- The pipeline buff list head
newst -- New pipeline buff to be inserted
-
static inline void wq_plb_queue_tail(wq_plb_list_t *list, wq_plb_t *newst)
Queue new packet in the tail of list.
- 参数:
list -- The pipeline buff list head
newst -- New pipeline buff to be inserted
-
static inline void wq_plb_list_splice_head(const wq_plb_list_t *list, wq_plb_list_t *head)
Join first list into the head of second list.
- 参数:
list -- The pipeline buff list consumed
head -- The pipeline buff list modified
-
static inline void wq_plb_list_splice_tail(const wq_plb_list_t *list, wq_plb_list_t *head)
Join first list into the tail of second list.
- 参数:
list -- The pipeline buff list consumed
head -- The pipeline buff list modified
-
static inline void wq_plb_unlink(wq_plb_t *adb, wq_plb_list_t *list)
Remove a pipeline buff from list.
备注
This function must be called atomically
- 参数:
adb -- Pipeline buff to be unlinked
list -- The pipeline buff list head
-
static inline wq_plb_t *wq_plb_dequeue_head(wq_plb_list_t *list)
Find & dequeue the first pipeline buff from list.
- 参数:
list -- The pipeline buff list head
- 返回:
First pipeline buff in the list
-
static inline wq_plb_t *wq_plb_dequeue_tail(wq_plb_list_t *list)
Find & dequeue the last pipeline buff from list.
- 参数:
list -- The pipeline buff list head
- 返回:
Last pipeline buff in the list
-
static inline wq_plb_t *wq_plb_dequeue_all(wq_plb_list_t *list)
Dequeue all the pipeline buff from list.
- 参数:
list -- The pipeline buff list head
- 返回:
First pipeline buff in the list
-
static inline void wq_plb_get(wq_plb_t *adb)
Add users for the pipeline buff.
- 参数:
adb -- The current pipeline buff
-
static inline uint32_t wq_plb_get_users(wq_plb_t *adb)
Get number of users belonged to pipeline buff.
- 参数:
adb -- The current pipeline buff
- 返回:
Number of users belonged to this pipeline buff
Check if the pipeline buff is shared by users.
- 参数:
adb -- The current pipeline buff
- 返回:
True if the pipeline buff is shared
-
static inline uint32_t wq_plb_get_channels(wq_plb_t *adb)
Get number of data channels in pipeline buff.
- 参数:
adb -- The current packet
- 返回:
Number of data channels in pipeline buff
-
wq_plb_t *wq_plb_alloc_header(void)
Alloc a pipeline buff head without buffer.
- 返回:
Alloced pipeline buff head
-
wq_plb_t *wq_plb_alloc(uint32_t size, uint32_t pad)
Alloc a pipeline buff head with buffer.
- 参数:
size -- The request buffer size
pad -- The preset pad size before data
- 返回:
Alloced pipeline buff
-
wq_plb_t *wq_plb_clone(wq_plb_t *adb)
Alloc a pipeline buff head and point it to the same data in old pipeline buff.
- 参数:
adb -- The pipeline buff to be cloned
- 返回:
Pointer of cloned pipeline buff
-
wq_plb_t *wq_plb_copy(wq_plb_t *adb)
Alloc a pipeline buff and copy data from old pipeline buff to it.
- 参数:
adb -- The pipeline buff to be copied
- 返回:
Pointer of copied pipeline buff
-
void wq_plb_free_list(wq_plb_list_t *list)
Free a pipeline buff list.
- 参数:
list -- The pipeline buff list to be free
-
struct wq_plb
Public Members
-
uint8_t *buf
Start pointer of data buffer belonged to this pipeline buff NULL for no buffer alloced
-
uint8_t *end
End pointer of data buffer, NULL for data length should be hold
-
uint8_t *data
Pointer of data start
-
uint32_t sp
Sample point in this pipeline buff
-
uint32_t len
Length (in bytes) of one channel data for multi-channel packet
-
uint32_t channel_gap
Gap (in bytes) from one channel data to another channel
-
uint32_t ts
Main path time stamp of pipeline data
-
uint32_t ts_sec
Tone path time stamp of pipeline data
-
uint32_t sample_rate
Data sample rate
-
uint32_t flags
Flags of this pipeline buff
-
uint8_t format
Format of pipeline buff data, e.g., pcm, aac, msbc
-
uint8_t channel
Channel bitmap for pipeline buff data
-
uint8_t dev_form
Audio production form
-
uint8_t bit_width
Sample bit width for pipeline buff data
-
uint8_t reserved
Reserved param
-
uint32_t users
Users of this pipeline buff
-
void *procedure
Reserved for pipeline pipeline, pointer of pend procedure
-
void (*free_cb)(struct wq_plb*, void*)
When this pipeline buff is free, this callback function is called
-
void *free_cb_param
Parameter used by free callback function
-
uint16_t plb_id
Plb data which is associated to processor id and stream id consist by (process id << 8) | stream id
-
uint32_t param
User-defined parameter
-
uint8_t *buf
-
struct wq_plb_list_t
Defines
-
WQ_PIPELINE_GATHER_MAX_NUM
Max gather number in system
-
WQ_PIPELINE_GATHER_MAX_INPUT
Max output stream number for gather processor
-
struct wq_pipeline_gather_config_t
- #include <pipeline_gather.h>
Gather processor handler multi-input data, the data should have same sample bit-width and same sample rate.
Public Members
-
uint16_t stream_id
Id of stream that gather belonged
-
uint16_t bit_width
Sample bit width, only support 16bit/32bit
-
uint16_t depth
Depth of hold stb queue, set 0 to use default value
-
uint8_t gather_id
Id of used gather
-
bool primary
True to set this stream as gather primary stream
-
bool match_ts
True to match ts before run gather operation
-
bool use_sec_ts
True to use second timestamp instead of main timestamp
-
bool redirect_drop
True to drop data when rx_list depth is full, just for non-primary stream
-
wq_pipeline_processor_t *sub_processor
Sub processor called by gather processor
-
void *sub_processor_config
Sub processor configure
-
uint16_t stream_id
-
struct wq_pipeline_gather_stereo_config_t
Defines
-
WQ_PIPELINE_MIRROR_MAX_NUM
Max mirror number in system
-
WQ_PIPELINE_MIRROR_MAX_OUTPUT
Max output stream number for mirror processor
-
struct wq_pipeline_mirror_config_t
Public Members
-
uint16_t stream_id
Id of stream which called the mirror processor
-
uint16_t mirror_id
Id of used mirror
-
uint16_t depth
Depth of hold stb queue, set 0 to use default value
-
bool copy_mode
True to copy stb to dispatch
-
bool redirect_mode
True to drop output stb for current procedure after mirror
-
uint32_t bypass_id_map
Bypass stream id bitmap
-
bool dummy_trigger
-
uint16_t stream_id
Typedefs
-
typedef struct pipeline_processor_info wq_pipeline_processor_info_t
-
typedef struct pipeline_processor_preset wq_pipeline_processor_preset_t
-
typedef struct pipeline_processor wq_pipeline_processor_t
-
struct pipeline_processor_info
-
struct pipeline_processor_preset
Public Members
-
void *scratch
Scratch buffer point
-
uint16_t stream_id
Stream id
-
wq_pipeline_processor_info_t *info
Processor info point
-
void *scratch
-
struct pipeline_processor
Public Members
-
char *name
Processor name
-
WQ_RET (*query)(void*, struct pipeline_processor_info*)
Query processor & return processor info
-
void *(*probe)(void*, struct pipeline_processor_preset*)
Probe processor & return handler
-
void (*remove)(void*)
Remove processor handler
-
void (*reset)(void*)
Reset processor handler
-
char *name
-
struct wq_pipeline_shaper_config_t
Defines
-
CONFIG_PIPELINE_STREAM_MAX_NUM
Max pipeline stream num
-
CONFIG_PIPELINE_PRESET_TASK_MAX_NUM
Max pipeline preset task num
-
WQ_PIPELINE_STREAM_SCHEDULE_START
Pipeline stream framework. processor means struct pipeline_processor point, config means config info point for processor init. when processor param is NULL, means the stream schedule is end.
schedule end must be {NULL, NULL}.
-
WQ_PIPELINE_STREAM_SCHEDULE_STEP(processor, config)
-
WQ_PIPELINE_STREAM_SCHEDULE_END
Typedefs
-
typedef struct stream_schedule wq_pipeline_stream_schedule_t
Enums
-
enum WQ_PIPELINE_STREAM_CTX
Values:
-
enumerator WQ_PIPELINE_STREAM_CTX_OWN_TASK
Stream has own task, can be triggered
-
enumerator WQ_PIPELINE_STREAM_CTX_PRESET_TASK
Reserved. Stream belongs to preset task, can be triggered
-
enumerator WQ_PIPELINE_STREAM_CTX_IRQ_HANDLER
Reserved. Stream work can be triggered by irq handler
-
enumerator WQ_PIPELINE_STREAM_CTX_TIMER_HANDLER
Reserved. Stream work can be triggered by timer handler
-
enumerator WQ_PIPELINE_STREAM_CTX_HIGH_SHARE_TASK
Reserved. Stream belongs to high share task, can be triggered
-
enumerator WQ_PIPELINE_STREAM_CTX_LOW_SHARE_TASK
Reserved. Stream belongs to low share task, can be triggered
-
enumerator WQ_PIPELINE_STREAM_CTX_CALLED
Stream work can only be called, can not be triggered
-
enumerator WQ_PIPELINE_STREAM_CTX_OWN_TASK
-
enum WQ_PIPELINE_STREAM_COMMAND
Values:
-
enumerator WQ_PIPELINE_STREAM_COMMAND_INVALID
Invalid command flag
-
enumerator WQ_PIPELINE_STREAM_COMMAND_CREATE
Command to create a stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_DESTROY
Command to destroy a stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_PEND
Command to pend a stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_RESUME
Command to resume a stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_QUERY
Command to query a stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_RESET_PROCEDURE
Command to reset a procedure in stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_BYPASS_PROCEDURE
Command to bypass a procedure in stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_UPDATE_PROCEDURE
Command to update runtime parameter of a procedure in stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_DEBUG_PROCEDURE
Command to enable log dump for a procedure in stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_DUMP_PROCEDURE_INPUT
Command to enable input plb dump for a procedure in stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_DUMP_PROCEDURE_OUTPUT
Command to enable output plb dump for a procedure in stream
-
enumerator WQ_PIPELINE_STREAM_COMMAND_MAX
Command max flag
-
enumerator WQ_PIPELINE_STREAM_COMMAND_INVALID
Functions
-
WQ_RET wq_pipeline_stream_control(wq_pipeline_stream_command_t *command)
Send control command to pipeline stream.
- 参数:
command -- The command to be send to pipeline stream.
- 返回:
WQ_RET_OK success.
WQ_RET_INVAL command is invalid.
-
struct stream_schedule
Public Members
-
const wq_pipeline_processor_t *processor
Processor point
-
void *config
Pipeline stream config info point
-
const wq_pipeline_processor_t *processor
-
struct wq_pipeline_stream_task_cfg_t
-
struct wq_pipeline_stream_preset_task_t
Public Members
-
uint16_t preset_task_id
Preset task id
-
bool preset_task_perpetuate
Preset task perpetuate attribute
-
wq_pipeline_stream_task_cfg_t task_cfg
Preset task config
-
uint16_t preset_task_id
-
struct wq_pipeline_stream_task_handle_t
-
struct wq_pipeline_stream_runner_t
Public Members
-
WQ_PIPELINE_STREAM_CTX type
Stream ctx type
-
void *config
Stream task config point
-
struct stream_schedule *schedule
Stream schedule
-
void *scratch
Stream scratch memory point
-
uint32_t scratch_size
Stream scratch memory size
-
uint32_t run_times
Stream tigger run times, will decrease one when each run loop
-
WQ_PIPELINE_STREAM_CTX type
-
struct wq_pipeline_stream_command_t
Public Members
-
struct list_head node
Command list node
-
WQ_PIPELINE_STREAM_COMMAND type
Command type
-
wq_pipeline_processor_t *processor
Which processor to handle the command
-
uint16_t procedure_index
Procedure index of stream
-
uint8_t stream_id
Stream id of the stream to handle the command
-
bool stream_depend
The command is depend stream
-
uint32_t init_sample_rate
Init sample rate for create command
-
void *callback_param
The param for callback handler
-
void (*callback)(void *callback_param)
Callback handler
-
void *param
Command param
-
struct list_head node