Die Leistungsergebnisse unterscheiden sich zwischen run_in_threadpool() und run_in_executor() in FastAPI ⇐ Python

[Anonymous]

Hier ist ein minimal reproduzierbares Beispiel meiner FastAPI-App. Ich habe ein seltsames Verhalten und bin mir nicht sicher, ob ich den Grund verstehe.
Ich verwende ApacheBench (

Code: Select all

ab

), um mehrere Anfragen wie folgt zu senden:

Code: Select all

ab -n 1000 -c 50 -H 'accept: application/json' -H 'x-data-origin: source' 'http://localhost:8001/test/async'

FastAPI-App

Code: Select all

import time
import asyncio
import enum
from typing import Any

from fastapi import FastAPI, Path, Body
from starlette.concurrency import run_in_threadpool

app = FastAPI()
loop = asyncio.get_running_loop()
def sync_func() -> None:
time.sleep(3)
print("sync func")

async def sync_async_with_fastapi_thread() -> None:
await run_in_threadpool( time.sleep, 3)
print("sync async with fastapi thread")

async def sync_async_func() -> None:
await loop.run_in_executor(None, time.sleep, 3)

async def async_func() -> Any:
await asyncio.sleep(3)
print("async func")

@app.get("/test/sync")
def test_sync() -> None:
sync_func()
print("sync")

@app.get("/test/async")
async def test_async() -> None:
await async_func()
print("async")

@app.get("/test/sync_async")
async def test_sync_async() -> None:
await sync_async_func()
print("sync async")

@app.get("/test/sync_async_fastapi")
async def test_sync_async_with_fastapi_thread() -> None:
await sync_async_with_fastapi_thread()
print("sync async with fastapi thread")

Hier sind die ApacheBench-Ergebnisse:
async with (asyncio.sleep) :
*Parallelitätsstufe: 50

• Testzeit: 63,528 Sekunden
• Vollständige Anfragen: 1000
  < li>Fehlgeschlagene Anfragen: 0
• Gesamtübertragung: 128000 Bytes
• HTML-Übertragung: 4000 Bytes
• Anfragen pro Sekunde: 15,74 [ #/sec] (Mittelwert)
• Zeit pro Anfrage: 3176,407 [ms] (Mittelwert)
• Zeit pro Anfrage : 63,528 [ms] (Mittelwert über alle gleichzeitigen Anfragen)
  Übertragungsrate: 1,97 [Kbytes/Sek.] empfangen*

sync (mit time.sleep):
Parallelitätsgrad: 50

• *Testzeit: 78,615 Sekunden
• Abgeschlossene Anfragen: 1000
• Fehlgeschlagene Anfragen: 0
• Gesamtübertragung: 128000 Bytes
• Übertragenes HTML: 4000 Bytes
• Anfragen pro Sekunde: 12,72 [#/s] (Mittelwert)
• Zeit pro Anfrage: 3930,751 [ms] (Mittelwert)
• Zeit pro Anfrage: 78,615 [ms] (Mittelwert, über alle gleichzeitigen Anforderungen hinweg)
  Übertragungsrate: 1,59 [Kbytes/Sek ] empfangen*

sync_async (Zeitschlaf mit run_in_executor) : *Parallelität Level: 50

• Testzeit: 256,201 Sekunden
• Abgeschlossene Anfragen: 1000
• Fehlgeschlagene Anfragen: 0
• Gesamtübertragung: 128.000 Bytes
• Übertragenes HTML: 4.000 Bytes
• Anfragen pro Sekunde: 3,90 [#/sec] (Mittelwert)
• Zeit pro Anfrage: 12810,038 [ms] (Mittelwert)
• Zeit pro Anfrage: 256,201 [ms] (Mittelwert über alle gleichzeitigen Anfragen)
  Übertragungsrate: 0,49 [Kbytes/Sek.] empfangen*

sync_async_fastapi (Zeitschlaf mit run_in Threadpool):
*Parallelitätsstufe: 50

• Benötigte Zeit für Tests: 78,877 Sekunden
• Abgeschlossene Anfragen: 1000
• Fehlgeschlagene Anfragen: 0
• Gesamtübertragung: 128000 Bytes
• Übertragenes HTML: 4000 Bytes
• Anfragen pro Sekunde: 12,68 [#/sec] (Mittelwert)
  < li>Zeit pro Anfrage: 3943,841 [ms] (Mittelwert)
• Zeit pro Anfrage: 78,877 [ms] (Mittelwert, über alle gleichzeitigen Prozesse hinweg). Anfragen)
  Übertragungsrate: 1,58 [Kbytes/Sek.] empfangen*

Zusammenfassend lässt sich sagen, dass ich insbesondere bei den Ergebnissen eine überraschende Ungleichheit erlebe bei Verwendung von run_in_executor, bei dem ich auf deutlich höhere Durchschnittszeiten stoße (12 Sekunden). Ich verstehe dieses Ergebnis nicht.
--- EDIT ---
Nach AKX-Antwort.

Code: Select all

Here the code working as expected:
import time
import asyncio
from anyio import to_thread

to_thread.current_default_thread_limiter().total_tokens = 200
loop = asyncio.get_running_loop()
executor = ThreadPoolExecutor(max_workers=100)
def sync_func() -> None:
time.sleep(3)
print("sync func")

async def sync_async_with_fastapi_thread() -> None:
await run_in_threadpool( time.sleep, 3)
print("sync async with fastapi thread")

async def sync_async_func() -> None:
await loop.run_in_executor(executor, time.sleep, 3)

async def async_func() -> Any:
await asyncio.sleep(3)
print("async func")

@app.get("/test/sync")
def test_sync() -> None:
sync_func()
print("sync")

@app.get("/test/async")
async def test_async() -> None:
await async_func()
print("async")

@app.get("/test/sync_async")
async def test_sync_async() -> None:
await sync_async_func()
print("sync async")

@app.get("/test/sync_async_fastapi")
async def test_sync_async_with_fastapi_thread() -> None:
await sync_async_with_fastapi_thread()
print("sync async with fastapi thread")