Run image decoding and resizing off the event loop
process_image takes ~560ms on a 12MP photo and image_content_block decodes HEIC/JPEG with PIL; both ran synchronously inside async handlers, freezing every concurrent chat stream. Run them via asyncio.to_thread in the upload handlers, the chat stream, and iMessage. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -257,7 +257,9 @@ async def upload_image():
|
||||
content_type = file.content_type or "image/jpeg"
|
||||
|
||||
try:
|
||||
processed_bytes, output_content_type = process_image(file_bytes, content_type)
|
||||
processed_bytes, output_content_type = await asyncio.to_thread(
|
||||
process_image, file_bytes, content_type
|
||||
)
|
||||
except ImageValidationError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
|
||||
@@ -311,8 +313,8 @@ async def upload_attachment():
|
||||
|
||||
if content_type in IMAGE_TYPES:
|
||||
try:
|
||||
processed_bytes, output_content_type = process_image(
|
||||
file_bytes, content_type
|
||||
processed_bytes, output_content_type = await asyncio.to_thread(
|
||||
process_image, file_bytes, content_type
|
||||
)
|
||||
except Exception as e:
|
||||
return jsonify({"error": f"Could not process image: {e}"}), 400
|
||||
@@ -402,7 +404,7 @@ async def stream_query():
|
||||
if image_key:
|
||||
try:
|
||||
image_bytes, _ = await s3_get_image(image_key)
|
||||
image_block = image_content_block(image_bytes)
|
||||
image_block = await asyncio.to_thread(image_content_block, image_bytes)
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to load image {image_key}: {e}")
|
||||
query_text = f"{query_text or ''}\n\n[An image was sent but could not be loaded.]".strip()
|
||||
|
||||
@@ -327,7 +327,7 @@ async def _process_and_reply(
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.get(media_url, timeout=30)
|
||||
resp.raise_for_status()
|
||||
image_block = image_content_block(resp.content)
|
||||
image_block = await asyncio.to_thread(image_content_block, resp.content)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load iMessage image: {e}")
|
||||
content = f"{content or ''}\n\n[An image was sent but could not be loaded.]".strip()
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Tests for conversation file attachments: extraction, /upload and stream-query file_key."""
|
||||
|
||||
import io
|
||||
import threading
|
||||
from datetime import timedelta
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, patch
|
||||
@@ -190,6 +191,21 @@ class TestUploadRoute:
|
||||
s3.upload_image.assert_awaited_once()
|
||||
s3.upload_file.assert_not_awaited()
|
||||
|
||||
async def test_image_processing_runs_off_the_event_loop(
|
||||
self, app, headers, s3, monkeypatch
|
||||
):
|
||||
threads = []
|
||||
real = conv.process_image
|
||||
|
||||
def recording(*args):
|
||||
threads.append(threading.current_thread())
|
||||
return real(*args)
|
||||
|
||||
monkeypatch.setattr(conv, "process_image", recording)
|
||||
resp = await _upload(app, headers, _png_bytes(), "cat.png", "image/png")
|
||||
assert resp.status_code == 200
|
||||
assert threads and threads[0] is not threading.main_thread()
|
||||
|
||||
async def test_unsupported_type_is_415(self, app, headers, s3):
|
||||
resp = await _upload(app, headers, b"PK\x03\x04", "a.zip", "application/zip")
|
||||
assert resp.status_code == 415
|
||||
|
||||
Reference in New Issue
Block a user