[Python] Unexpected latency FastAPI and asyncio in real time - WebSocket

I am currently building an API for real-time processing of any kind of data (files/audios/images). Regardless of the file type, the API will receive the same quantity of bytes. The goal is clearly the responsiveness of the API.

The API is rather simple


from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
import uvicorn
import time


app = FastAPI()

def cpu_bound_task():
# contains CPU-bound task, but really fast ~10ms or less
return

async def test():
await asyncio.sleep(0.01)



@app.get("/", status_code=200)
async def ping():
return {"message": "working well !"}


@app.websocket("/stream")
async def streaming(websocket: WebSocket):
await websocket.accept()
start_connection = time.time()



try:
while True:
data = await websocket.receive_bytes()
#do some CPU-bound task
cpu_bound_task()
# or using test() function

if data == b"\x00":
await websocket.send_text(str(time.time() - start_connection))
break # EOF received

except WebSocketDisconnect:
print("Client disconnected before sending EOF")

finally:
pass


I wrote 2 specific functions : cpu_bound_task is not asynchronous because as stated in this answer I think I need async for a CPU-bound task as it will be blocking even for the event_loop and avoiding async will spawn it on a different thread. And also a function to reproduce the cpu_bound_task, test.

To simulate some client requests i used the following script

async def stream_audio(uri, my_file):
async with websockets.connect(uri) as websocket:
with open(my file, "rb") as wf:
while True:
chunk = wf.read(CHUNK_SIZE)
if not chunk:
break
await websocket.send(chunk)
await asyncio.sleep(0.075) # because we recieve 75ms by 75ms
await websocket.send(b"\x00")
result = await websocket.recv()
await websocket.close()

return {my_file: float(result)}


async def run_multiple_clients(files, concurrent_clients=5):
tasks = []

for i in range(concurrent_clients):
path = files[i % len(files)]
tasks.append(stream_audio(URI, path))

await asyncio.gather(*tasks)



What I am having hard to understand is when I will pass let's say a single audio file of 20 seconds by chunk of 75ms with a pause of the same size to reproduce real-time, the processing time will have an offset of around 200-400ms even if I am doing nothing in the API.

• I know the internal clock of asyncio is not entirely precise
• I know that CPU-bound task can be blocking but it should be fine as long as the task takes less time than 75ms

As clearly unexperimented i am having a hard time understanding this offset/jitter of 200-400ms that even appear without CPU Bound tasks and asynchronous function like the following one.

async def test2():
await asyncio.sleep(0.01)


Or simply because I should use docker to isolate my FastAPI and ensure simulate_client.py is ran on another CPU

Continue reading...