Use thread-per-request for async UDFs instead of shared event loop

The dedicated shared event loop still caused starvation under concurrent
async UDF calls. Switch to the same model used by sync UDFs: each request
gets its own thread with asyncio.run(), eliminating loop contention.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: kesmit13

singlestoredb/apps/_python_udfs.py

@@ -10,9 +10,8 @@
 if typing.TYPE_CHECKING:
     from ._uvicorn_util import AwaitableUvicornServer
 
-# Keep track of currently running server and app
+# Keep track of currently running server
 _running_server: 'typing.Optional[AwaitableUvicornServer]' = None
-_running_app: typing.Optional[Application] = None
 
 # Maximum number of UDFs allowed
 MAX_UDFS_LIMIT = 10
@@ -22,7 +21,7 @@ async def run_udf_app(
     log_level: str = 'error',
     kill_existing_app_server: bool = True,
 ) -> UdfConnectionInfo:
-    global _running_server, _running_app
+    global _running_server
     from ._uvicorn_util import AwaitableUvicornServer
 
     try:
@@ -39,9 +38,6 @@ async def run_udf_app(
         if _running_server is not None:
             await _running_server.shutdown()
             _running_server = None
-        if _running_app is not None:
-            _running_app.shutdown()
-            _running_app = None
 
         # Kill if any other process is occupying the port
         kill_process_by_port(app_config.listen_port)
@@ -58,32 +54,27 @@ async def run_udf_app(
         log_level=log_level,
     )
 
-    try:
-        if not app.endpoints:
-            raise ValueError('You must define at least one function.')
-        if len(app.endpoints) > MAX_UDFS_LIMIT:
-            raise ValueError(
-                f'You can only define a maximum of {MAX_UDFS_LIMIT} functions.',
-            )
-
-        config = uvicorn.Config(
-            app,
-            host='0.0.0.0',
-            port=app_config.listen_port,
-            log_config=app.get_uvicorn_log_config(),
+    if not app.endpoints:
+        raise ValueError('You must define at least one function.')
+    if len(app.endpoints) > MAX_UDFS_LIMIT:
+        raise ValueError(
+            f'You can only define a maximum of {MAX_UDFS_LIMIT} functions.',
         )
 
-        # Register the functions only if the app is running interactively.
-        if app_config.running_interactively:
-            app.register_functions(replace=True)
-
-        _running_app = app
-        _running_server = AwaitableUvicornServer(config)
-        asyncio.create_task(_running_server.serve())
-        await _running_server.wait_for_startup()
-    except Exception:
-        app.shutdown()
-        raise
+    config = uvicorn.Config(
+        app,
+        host='0.0.0.0',
+        port=app_config.listen_port,
+        log_config=app.get_uvicorn_log_config(),
+    )
+
+    # Register the functions only if the app is running interactively.
+    if app_config.running_interactively:
+        app.register_functions(replace=True)
+
+    _running_server = AwaitableUvicornServer(config)
+    asyncio.create_task(_running_server.serve())
+    await _running_server.wait_for_startup()
 
     print(f'Python UDF registered at {base_url}')
 

singlestoredb/functions/ext/asgi.py

@@ -24,7 +24,6 @@
 """
 import argparse
 import asyncio
-import concurrent.futures
 import contextvars
 import dataclasses
 import datetime
@@ -1001,9 +1000,6 @@ def __init__(
         self.log_level = log_level
         self.disable_metrics = disable_metrics
 
-        self._udf_loop: Optional[asyncio.AbstractEventLoop] = None
-        self._udf_thread: Optional[threading.Thread] = None
-
         # Configure logging
         self._configure_logging()
 
@@ -1037,27 +1033,6 @@ def _configure_logging(self) -> None:
         # Prevent propagation to avoid duplicate or differently formatted messages
         self.logger.propagate = False
 
-    def _get_udf_loop(self) -> asyncio.AbstractEventLoop:
-        """Get or create the dedicated UDF event loop."""
-        if self._udf_loop is None:
-            self._udf_loop = asyncio.new_event_loop()
-            self._udf_thread = threading.Thread(
-                target=self._udf_loop.run_forever,
-                daemon=True,
-                name='async-udf-loop',
-            )
-            self._udf_thread.start()
-        return self._udf_loop
-
-    def shutdown(self) -> None:
-        """Shut down the dedicated UDF event loop."""
-        if self._udf_loop is not None:
-            self._udf_loop.call_soon_threadsafe(self._udf_loop.stop)
-        if self._udf_thread is not None:
-            self._udf_thread.join(timeout=5)
-        self._udf_loop = None
-        self._udf_thread = None
-
     def get_uvicorn_log_config(self) -> Dict[str, Any]:
         """
         Create uvicorn log config that matches the Application's logging format.
@@ -1206,7 +1181,6 @@ async def __call__(
             try:
                 all_tasks = []
                 result = []
-                udf_future: 'Optional[concurrent.futures.Future[Any]]' = None
 
                 cancel_event = threading.Event()
 
@@ -1215,23 +1189,13 @@ async def __call__(
                         func_info['colspec'], b''.join(data),
                     )
 
-                func_task: 'asyncio.Task[Any]'
-                if func_info['is_async']:
-                    udf_future = asyncio.run_coroutine_threadsafe(
-                        func(cancel_event, call_timer, *inputs),
-                        self._get_udf_loop(),
-                    )
-                    func_task = asyncio.ensure_future(
-                        asyncio.wrap_future(udf_future),
-                    )
-                else:
-                    func_task = asyncio.create_task(
-                        to_thread(
-                            lambda: asyncio.run(
-                                func(cancel_event, call_timer, *inputs),
-                            ),
+                func_task = asyncio.create_task(
+                    to_thread(
+                        lambda: asyncio.run(
+                            func(cancel_event, call_timer, *inputs),
                         ),
-                    )
+                    ),
+                )
                 disconnect_task = asyncio.create_task(
                     asyncio.sleep(int(1e9))
                     if ignore_cancel else cancel_on_disconnect(receive),
@@ -1248,23 +1212,18 @@ async def __call__(
                     )
 
                 await cancel_all_tasks(pending)
-                if func_task in pending and udf_future is not None:
+                if func_task in pending:
                     cancel_event.set()
-                    udf_future.cancel()
 
                 for task in done:
                     if task is disconnect_task:
                         cancel_event.set()
-                        if udf_future is not None:
-                            udf_future.cancel()
                         raise asyncio.CancelledError(
                             'Function call was cancelled by client disconnect',
                         )
 
                     elif task is timeout_task:
                         cancel_event.set()
-                        if udf_future is not None:
-                            udf_future.cancel()
                         raise asyncio.TimeoutError(
                             'Function call was cancelled due to timeout',
                         )
@@ -1327,9 +1286,7 @@ async def __call__(
                 await send(self.error_response_dict)
 
             finally:
-                if udf_future is not None:
-                    cancel_event.set()
-                    udf_future.cancel()
+                cancel_event.set()
                 await cancel_all_tasks(all_tasks)
 
         # Handle api reflection