Реал‑тайм стриминг на GPU: низкая латентность

				
					from fastapi import FastAPI, WebSocket
from sse_starlette.sse import EventSourceResponse
import asyncio, time, json, base64, collections

app = FastAPI()
HWM, LWM = 8, 4
CHUNK_MS = 20

async def gpu_infer(blocks: list[bytes]) -> list[bytes]:
    # Единичный вызов модели над 1–4 блоками: FP16/BF16, zero-copy
    await asyncio.sleep(0)  # имитация
    return blocks  # заглушка

async def process_stream(frame_iter):
    q = collections.deque()
    last_emit = 0
    async def producer():
        async for frame in frame_iter:      # PCM блоками 20–40 мс
            q.append(frame)
            if len(q) > HWM and DROP_OLD_CHUNKS:
                q.popleft()                 # backpressure: сброс старых блоков
            await asyncio.sleep(0)
    async def consumer():
        nonlocal last_emit
        while True:
            if not q:
                await asyncio.sleep(0.001); continue
            batch = []
            while q and len(batch) < 4:     # микробатчинг до 4 блоков
                batch.append(q.popleft())
            out = await gpu_infer(batch)
            now = time.time()
            for chunk in out:
                if (now - last_emit) * 1000 >= 100:  # PARTIALS_EVERY_MS
                    last_emit = now
                    yield base64.b64encode(chunk).decode("ascii")
            await asyncio.sleep(0)
    return producer, consumer

@app.post("/sse")
async def sse_endpoint():
    async def event_gen():
        producer, consumer = await process_stream(audio_source())
        asyncio.create_task(producer())
        async for b64 in consumer():
            yield {"event": "audio", "data": b64}
        yield {"event": "done", "data": "{}"}
    return EventSourceResponse(event_gen())

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    await ws.accept()
    producer, consumer = await process_stream(websocket_audio_source(ws))
    asyncio.create_task(producer())
    async for b64 in consumer():
        await ws.send_text(b64)
    await ws.close()

# Заглушки источников аудио:
async def audio_source():
    while True:
        frame = await next_pcm_frame_or_none(CHUNK_MS)
        if frame is None: break
        yield frame

async def websocket_audio_source(ws: WebSocket):
    while True:
        msg = await ws.receive_text()
        yield base64.b64decode(msg)