使用trio调用cvlc播放网络电台的卡顿与_FdHolder报错问题

1. 退出报错原因：_FdHolder.__del__中的报错是因为cvlc进程终止后，stdin管道未被正确关闭，导致trio在清理资源时调用了已被置为None的函数。全局变量的异步竞态也会加剧这个问题。
2. 音频卡顿原因：
   • 循环内反复创建httpx.AsyncClient，额外开销导致拉流延迟
   • 用trio.sleep(1)和wait_writable做同步，时机不准确，导致流数据堆积或断供
   • 错误处理中pass掉BrokenResourceError，可能导致无效的写入尝试占用资源

一、修复退出报错 #

1. 正确清理进程资源：终止cvlc后，需等待进程结束并关闭stdin管道，避免资源泄漏
2. 替换全局变量为trio同步原语：用trio.Event和trio.Nursery的取消机制替代全局字典，避免异步竞态
3. 移除subprocess.PIPE依赖：直接使用trio原生的管道处理逻辑

二、解决音频卡顿 #

1. 复用httpx.AsyncClient：将client创建移到循环外，减少连接开销
2. 用事件同步替代sleep：确保receiver在cvlc启动完成后再开始拉流
3. 优化流处理逻辑：减少数据累积，及时处理音频块，避免内存占用过高
4. 完善错误处理：遇到管道错误时立即终止拉流，避免无效操作

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" Test """

import re
import trio
import httpx

# 用trio同步原语替代全局字典
class State:
    def __init__(self):
        self.player = None
        self.stop_event = trio.Event()
        self.player_ready = trio.Event()

CH = 'http://icecast.ndr.de/ndr/ndr2/niedersachsen/mp3/128/stream.mp3'

async def main():
    state = State()
    async with trio.open_nursery() as nursery:
        nursery.start_soon(play_audio, state, nursery)
        nursery.start_soon(receiver, state)

async def play_audio(state, nursery):
    cmd = ['cvlc', '--quiet', '-']
    # 使用trio的PIPE（默认就是，无需导入subprocess.PIPE）
    state.player = await trio.lowlevel.open_process(cmd, stdin=trio.lowlevel.PIPE)
    state.player_ready.set()  # 通知receiver播放器已就绪
    
    await trio.sleep(20)
    state.stop_event.set()  # 触发停止信号
    
    # 正确清理资源：先关闭stdin，再终止进程，最后等待结束
    await state.player.stdin.aclose()
    state.player.terminate()
    await state.player.wait()
    nursery.cancel_scope.cancel()  # 取消所有子任务

async def receiver(state):
    await state.player_ready.wait()  # 等待播放器就绪，替代sleep(1)
    btitle = b'StreamTitle='
    pos = 0
    data = b''
    # 复用同一个AsyncClient，避免反复创建
    async with httpx.AsyncClient() as client:
        while not state.stop_event.is_set():
            try:
                async with client.stream(
                    "GET", CH, 
                    headers={'Icy-MetaData': '1'},
                    follow_redirects=True
                ) as radio:
                    encoding = radio.headers.get('encoding', 'utf-8')
                    metaint = int(radio.headers.get('icy-metaint', '0'))
                    stream = radio.stream
                    async for chunk in stream:
                        if state.stop_event.is_set():
                            await stream.aclose()
                            break
                        
                        data += chunk
                        # 处理音频块和元数据
                        while pos + metaint <= len(data):
                            # 写入音频数据到播放器
                            try:
                                await state.player.stdin.send_all(data[pos:pos+metaint])
                            except trio.BrokenResourceError:
                                # 播放器已关闭，直接退出
                                await stream.aclose()
                                state.stop_event.set()
                                return
                            
                            pos += metaint
                            # 处理元数据
                            if pos >= len(data):
                                break
                            metalen = data[pos] * 16
                            pos += 1
                            if pos + metalen > len(data):
                                break
                            meta_data = data[pos:pos+metalen].rstrip(b'\0')
                            if meta_data.startswith(btitle):
                                title_match = re.search(br"StreamTitle='([^']*)';", meta_data)
                                if title_match:
                                    title = title_match.group(1).decode(encoding, errors='replace')
                                    print(f"Title: '{title}'")
                            pos += metalen
                        # 裁剪剩余未处理的数据
                        data = data[pos:]
                        pos = 0
            except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.ConnectError):
                # 重连前短暂等待，避免频繁重试
                await trio.sleep(2)
                continue

if __name__ == "__main__":
    trio.run(main)

1. 同步原语替代全局变量：用State类封装状态，stop_event和player_ready确保任务间同步安全，避免竞态
2. 资源清理：play_audio中先关闭stdin，再终止进程并等待结束，彻底释放管道资源，解决退出报错
3. Client复用：httpx.AsyncClient在receiver循环外创建，减少连接建立开销
4. 同步优化：用player_ready事件替代sleep(1)，确保播放器就绪后再拉流，避免初始断流
5. 错误处理：遇到BrokenResourceError直接退出，避免无效写入操作，减少资源占用
6. 流处理优化：调整元数据处理的边界判断，避免索引越界，同时及时裁剪已处理的数据，减少内存占用

内容的提问来源于stack exchange，提问作者Peter