import multiprocessing.shared_memory
import numpy as np

async def process_chunk(self, audio_array: np.ndarray, is_final: bool) -> str:
    audio_size = audio_array.nbytes

    # 1. 创建共享内存 (Zero-Copy 基础)
    shm = multiprocessing.shared_memory.SharedMemory(
        create=True,
        size=audio_size,
        name=f"asr_{task_id}"
    )

    # 2. 写入数据
    audio_bytes = audio_array.tobytes()
    shm.buf[:audio_size] = audio_bytes

    # 3. 引用管理（防止被过早 GC）
    async with self.factory._shared_memory_lock:
        self.factory._shared_memory_pool[shm.name] = shm

    # 4. 仅发送“指针”
    task = {
        "task_id": task_id,
        "shm_name": shm.name,
        "audio_size": audio_size,
        # ... 其他元数据
    }
    self._task_queue.put(task)

def worker_process():
    # ... 获取任务 ...
    if shm_name and audio_size > 0:
        # 直接连接已存在的共享内存
        shm = multiprocessing.shared_memory.SharedMemory(name=shm_name)
        # 从 buffer 重构 numpy 数组，几乎无开销
        audio_array = np.frombuffer(shm.buf[:audio_size], dtype=np.float32).copy()
        shm.close()
    # ... 模型推理 ...

# 限制最大并发连接数，超过直接拒绝或等待
_streaming_semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)

@router.websocket("/streaming")
async def transcription(websocket: WebSocket):
    try:
        # 快速失败机制：等待超时则拒绝服务
        await asyncio.wait_for(semaphore.acquire(), timeout=REQUEST_TIMEOUT)
    except asyncio.TimeoutError:
        await websocket.close(code=1008, reason="服务繁忙")
        return
    # ... 业务逻辑 ...

import anyio

# 创建有界内存流，防止内存无限增长
send_stream, receive_stream = anyio.create_memory_object_stream(max_buffer_size=100)

async with anyio.create_task_group() as tg:
    # 消费者：后台模型处理
    tg.start_soon(audio_processor_task, receive_stream)

    # 生产者：WebSocket 接收
    async with send_stream:
        while True:
            msg = await websocket.receive()
            await send_stream.send(audio_message)

# 典型的异步避坑：不要在 async def 里直接写 open()
def _write_file():
    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        tmp.write(content)
        return tmp.name

tmp_path = await loop.run_in_executor(None, _write_file)

@staticmethod
def select_from_factories(factories: List["ASRFactory"]) -> "ASRFactory":
    # 贪心算法：选择积压任务最少的 Worker 组
    return min(factories, key=lambda f: len(f._pending_tasks))