perf(backend): offload blocking calls in async paths to threads (#26381)

Audit of asyncio.sleep vs time.sleep and event-loop-blocking calls:

- utils/plugin.py: run pip `install_frontmatter_requirements`
  (subprocess.check_call) via asyncio.to_thread in load_tool_module_by_id,
  load_function_module_by_id, and install_tool_and_function_dependencies.
- retrieval/utils.py: move the synchronous SSRF-guarded requests probe and
  loader.load() in get_content_from_url into a sync helper run via
  asyncio.to_thread.
- routers/audio.py: write uploaded audio to disk off the event loop in
  transcription().
- routers/pipelines.py: write uploaded pipeline file off the event loop in
  upload_pipeline().

The existing time.sleep call sites are all in genuinely synchronous
functions (sync requests/DB drivers/daemon threads) with async
counterparts that already use asyncio.sleep, so no time.sleep -> asyncio.sleep
changes were needed.


Claude-Session: https://claude.ai/code/session_01LXR5bYfsfSS42RGHQZu2Ta

Co-authored-by: Claude <noreply@anthropic.com>

backend/open_webui/retrieval/utils.py

@@ -191,11 +191,17 @@ def _is_text_content_type(content_type: str) -> bool:
 
 
 async def get_content_from_url(request, url: str) -> str:
-    from open_webui.retrieval.web.utils import validate_url, _SSRFSafeAdapter
-
-
     loader_config = await get_loader_config()
 
+    # The rest of this function performs synchronous, blocking work: an SSRF-guarded
+    # `requests` probe and a synchronous document loader (`loader.load()`). Run it in a
+    # worker thread so the event loop stays free while waiting on network/parsing.
+    return await asyncio.to_thread(_get_content_from_url_sync, request, url, loader_config)
+
+
+def _get_content_from_url_sync(request, url: str, loader_config):
+    from open_webui.retrieval.web.utils import validate_url, _SSRFSafeAdapter
+
     # Validate URL before making any request (blocks private IPs, non-HTTP, filter list)
     validate_url(url)
 

backend/open_webui/routers/audio.py

@@ -1192,8 +1192,12 @@ async def transcription(
         if not os.path.realpath(file_path).startswith(os.path.realpath(file_dir)):
             raise ValueError('Invalid file path detected')
 
-        with open(file_path, 'wb') as f:
-            f.write(contents)
+        def _write_upload():
+            with open(file_path, 'wb') as f:
+                f.write(contents)
+
+        # Audio uploads can be large; write to disk off the event loop.
+        await asyncio.to_thread(_write_upload)
 
         try:
             metadata = None

backend/open_webui/routers/pipelines.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 import shutil
@@ -236,9 +237,12 @@ async def upload_pipeline(
 
     response = None
     try:
-        # Save the uploaded file
-        with open(file_path, 'wb') as buffer:
-            shutil.copyfileobj(file.file, buffer)
+        # Save the uploaded file off the event loop (uploads can be large).
+        def _save_upload():
+            with open(file_path, 'wb') as buffer:
+                shutil.copyfileobj(file.file, buffer)
+
+        await asyncio.to_thread(_save_upload)
 
         url, key = await get_openai_connection(urlIdx)
 

backend/open_webui/utils/plugin.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
 import re
@@ -213,8 +214,12 @@ async def load_tool_module_by_id(tool_id, content=None):
         await Tools.update_tool_by_id(tool_id, {'content': content})
     else:
         frontmatter = extract_frontmatter(content)
-        # Install required packages found within the frontmatter
-        install_frontmatter_requirements(frontmatter.get('requirements', ''))
+        # Install required packages found within the frontmatter.
+        # Runs `pip install` via subprocess, which can take a long time;
+        # offload to a thread so it doesn't block the event loop.
+        await asyncio.to_thread(
+            install_frontmatter_requirements, frontmatter.get('requirements', '')
+        )
 
     module_name = f'tool_{tool_id}'
     module = types.ModuleType(module_name)
@@ -258,7 +263,10 @@ async def load_function_module_by_id(function_id: str, content: str | None = Non
         await Functions.update_function_by_id(function_id, {'content': content})
     else:
         frontmatter = extract_frontmatter(content)
-        install_frontmatter_requirements(frontmatter.get('requirements', ''))
+        # `pip install` via subprocess can block for a long time; offload it.
+        await asyncio.to_thread(
+            install_frontmatter_requirements, frontmatter.get('requirements', '')
+        )
 
     module_name = f'function_{function_id}'
     module = types.ModuleType(module_name)
@@ -459,6 +467,9 @@ async def install_tool_and_function_dependencies():
                 if dependencies := frontmatter.get('requirements'):
                     all_dependencies += f'{dependencies}, '
 
-        install_frontmatter_requirements(all_dependencies.strip(', '))
+        # `pip install` via subprocess can block for a long time; offload it.
+        await asyncio.to_thread(
+            install_frontmatter_requirements, all_dependencies.strip(', ')
+        )
     except Exception as e:
         log.error(f'Error installing requirements: {e}')