Linux PipeWire深度解析之pw_thread_loop_timed_wait调用流程与实战(十九)

简介：CSDN博客专家、《Android系统多媒体进阶实战》作者

博主新书推荐：《Android系统多媒体进阶实战》🚀
Android Audio工程师专栏地址：Audio工程师进阶系列【原创干货持续更新中……】🚀
Android多媒体专栏地址：多媒体系统工程师系列【原创干货持续更新中……】🚀
专题一 二：AAOS车载系统+AOSP14系统攻城狮入门视频实战课🚀
专题三：Android14 Binder之HIDL与AIDL通信实战课🚀
专题四：Android15快速自定义与集成音效实战课🚀
专题五：Android15音频策略实战课🚀
专题六：Android15音频性能实战课(无声/杂音/断音/爆音实战案例)🚀

人生格言：人生从来没有捷径，只有行动才是治疗恐惧和懒惰的唯一良药.

🌻1. 前言

本篇目的：

Linux PipeWire 深度解析之pw_thread_loop_timed_wait调用流程与实战。

要点概括

• 核心功能：在持有pw_thread_loop锁的前提下，等待其他线程的通知，并限制本次等待的最长时间。
• 工作机制：函数将相对超时时间转换为绝对截止时间，进入底层条件变量等待；被pw_thread_loop_signal()唤醒时返回0，超时时返回ETIMEDOUT。
• 典型用途：等待 Core 同步完成、Registry 枚举结束、Stream 状态变化或异步请求结果，避免主线程无限期阻塞。

🌻2. 应用场景与用法

pw_thread_loop_timed_wait()是 PipeWire Thread Loop 的跨线程同步接口。

当应用线程向 PipeWire 异步提交操作后，通常不能立即得到最终结果。例如调用pw_core_sync()后，完成结果会在 Core 的done回调中异步到达。

此时应用可以：

pw_thread_loop_lock(thread_loop);

持有 Thread Loop 锁，发起异步请求后调用：

pw_thread_loop_timed_wait(thread_loop,3);

等待回调线程更新状态并通过：

pw_thread_loop_signal(thread_loop,false);

唤醒等待者。

与pw_thread_loop_wait()相比：

pw_thread_loop_wait()可无限等待；pw_thread_loop_timed_wait()为单次等待增加超时上限，调用者能够处理 PipeWire 服务不可用、回调未到达或业务状态异常等情况。

函数原型

intpw_thread_loop_timed_wait(structpw_thread_loop*loop,intwait_max_sec);

参数说明

loop:由pw_thread_loop_new()创建的 Thread Loop 对象 wait_max_sec:本次等待允许的最长秒数

调用该函数前，必须已经持有：

pw_thread_loop_lock(loop);

获取的锁。

返回值

0:被pw_thread_loop_signal()唤醒 ETIMEDOUT:等待时间到达上限 负数:底层等待过程发生错误

注意：

ETIMEDOUT 只表示本次等待结束， 不代表此前提交的异步 PipeWire 请求已经取消。

应用场景

pw_thread_loop_timed_wait()常见应用场景主要有三类。

第一类是等待 PipeWire Core 同步完成。应用调用pw_core_sync()后，需要等待done回调确认服务端已经处理到指定序列号；使用限时等待可以避免 Core 异常时永久卡住。

第二类是等待 Registry 枚举结果。客户端连接 Core 并注册 Registry Listener 后，节点、端口或设备信息会陆续通过回调到达。应用可在回调中记录状态并唤醒等待线程，再用超时处理枚举未完成的情况。

第三类是等待 Stream 状态切换。例如创建播放或录音 Stream 后，应用可能要等待其变为PW_STREAM_STATE_PAUSED或PW_STREAM_STATE_STREAMING。限时等待能让控制线程保留失败恢复与资源清理的机会。

🌻3. 调用流程剖析

🌻3.1 核心步骤

1. 获取 Thread Loop 锁

应用线程先调用：

pw_thread_loop_lock(thread_loop);

保护后续共享状态，例如：

同步是否完成 回调返回结果 请求序列号 超时标记

2. 发起异步 PipeWire 请求

以 Core 同步为例：

sequence=pw_core_sync(core,PW_ID_CORE,0);

该调用只提交同步请求，最终结果会通过 Core 的done回调异步返回。

3. 检查等待条件

应用不能假定调用一次等待函数后状态一定已经完成，而应先检查共享条件：

while(!sync_done){...}

这样可以应对虚假唤醒、多个等待者以及非目标事件带来的唤醒。

4. 调用 pw_thread_loop_timed_wait

应用执行：

result=pw_thread_loop_timed_wait(thread_loop,3);

请求在当前锁保护范围内等待，最长等待三秒。

5. 计算绝对超时时间

pw_thread_loop_timed_wait()根据当前时间和wait_max_sec计算绝对截止时间，然后进入内部的完整超时等待实现。

底层条件变量接口需要绝对时间，因此超时判断由条件变量等待机制完成。

6. 释放锁并等待条件变量

进入等待后，底层会原子地完成：

释放 Thread Loop 锁 进入条件变量等待

这样 PipeWire 工作线程才能取得同一把锁，执行回调并更新共享状态。

7. 回调更新状态并发送通知

PipeWire 工作线程执行 Core、Registry 或 Stream 回调。

回调在更新共享状态后调用：

pw_thread_loop_signal(thread_loop,false);

通知正在等待的线程。

8. 等待线程重新获得锁

等待线程被唤醒后，会先重新获取 Thread Loop 锁，再从：

pw_thread_loop_timed_wait(...);

返回。

因此调用者返回后仍处于锁保护范围，可安全读取回调写入的共享状态。

9. 处理唤醒或超时结果

如果返回：

0

应用重新检查等待条件；如果返回：

ETIMEDOUT

说明本次等待超时，应进入超时处理、错误上报或资源清理逻辑。

🌻3.2 调用流程图

🌻3.3 超时与唤醒时序图

🌻4. 实战应用案例

下面示例创建 Thread Loop 并连接 PipeWire Core，调用pw_core_sync()后通过pw_thread_loop_timed_wait()等待done回调。

#include<pipewire/pipewire.h>#include<errno.h>#include<stdbool.h>#include<stdio.h>#include<stdlib.h>structapp_data{structpw_thread_loop*thread_loop;structpw_context*context;structpw_core*core;structspa_hookcore_listener;intsequence;intresult;bool sync_done;};staticvoidon_core_done(void*data,uint32_tid,intseq){structapp_data*app=data;if(id!=PW_ID_CORE||seq!=app->sequence){return;}app->sync_done=true;app->result=0;pw_thread_loop_signal(app->thread_loop,false);}staticvoidon_core_error(void*data,uint32_tid,intseq,intres,constchar*message){structapp_data*app=data;fprintf(stderr,"PipeWire Core error: %s\n",message);app->sync_done=true;app->result=res;pw_thread_loop_signal(app->thread_loop,false);}staticconststructpw_core_eventscore_events={PW_VERSION_CORE_EVENTS,.done=on_core_done,.error=on_core_error,};intmain(intargc,char*argv[]){structapp_dataapp={.result=-EIO};intwait_result;intexit_code=EXIT_FAILURE;pw_init(&argc,&argv);app.thread_loop=pw_thread_loop_new("pipewire-timed-wait",NULL);if(!app.thread_loop){fprintf(stderr,"pw_thread_loop_new failed\n");gotodone;}app.context=pw_context_new(pw_thread_loop_get_loop(app.thread_loop),NULL,0);if(!app.context){fprintf(stderr,"pw_context_new failed\n");gotodestroy_loop;}app.core=pw_context_connect(app.context,NULL,0);if(!app.core){fprintf(stderr,"pw_context_connect failed\n");gotodestroy_context;}pw_core_add_listener(app.core,&app.core_listener,&core_events,&app);if(pw_thread_loop_start(app.thread_loop)<0){fprintf(stderr,"pw_thread_loop_start failed\n");gotodestroy_core;}pw_thread_loop_lock(app.thread_loop);app.sequence=pw_core_sync(app.core,PW_ID_CORE,0);if(app.sequence<0){fprintf(stderr,"pw_core_sync failed: %d\n",app.sequence);app.result=app.sequence;gotounlock_loop;}while(!app.sync_done){wait_result=pw_thread_loop_timed_wait(app.thread_loop,3);if(wait_result==ETIMEDOUT){fprintf(stderr,"wait for Core sync timed out\n");app.result=-ETIMEDOUT;break;}if(wait_result<0){fprintf(stderr,"pw_thread_loop_timed_wait failed: %d\n",wait_result);app.result=wait_result;break;}}if(app.result==0){exit_code=EXIT_SUCCESS;}unlock_loop:pw_thread_loop_unlock(app.thread_loop);pw_thread_loop_stop(app.thread_loop);destroy_core:spa_hook_remove(&app.core_listener);pw_core_disconnect(app.core);destroy_context:pw_context_destroy(app.context);destroy_loop:pw_thread_loop_destroy(app.thread_loop);done:pw_deinit();returnexit_code;}

该示例的关键点是：

• 主线程在持有 Thread Loop 锁时提交pw_core_sync()请求。
• done回调运行后写入sync_done，随后调用pw_thread_loop_signal()。
• 主线程从pw_thread_loop_timed_wait()返回后，重新检查sync_done。
• 超时只结束当前等待；若应用需要控制整体等待时长，应自行记录截止时间，或使用更底层的绝对时间等待接口。

🌻5. 一句话总结

pw_thread_loop_timed_wait()本质上是：

“在持有 PipeWire Thread Loop 锁时等待异步事件，并在限定时间内返回唤醒结果或超时结果。”