GitHub Viewer
"""Application execution on a standard asyncio event loop.
Mobile hosts start one application thread with :func:`start`. All Python
rendering, callbacks, and tasks run there. Native hosts own their UI threads
and marshal view operations themselves. Headless programs can instead drive
an ordinary local loop with :func:`run_blocking` and :func:`drain`.
"""
from __future__ import annotations
import asyncio
import contextvars
import inspect
import threading
from typing import Any, Awaitable, Callable, Coroutine, TypeVar
T = TypeVar("T")
Awaitlike = Coroutine[Any, Any, T] | Awaitable[T]
_loop: asyncio.AbstractEventLoop | None = None
_thread: threading.Thread | None = None
_owner: threading.Thread | None = None
_lock = threading.RLock()
_scope: contextvars.ContextVar[TaskScope | None] = contextvars.ContextVar("pn_task_scope", default=None)
class TaskScope:
"""Own tasks until an application or component is disposed.
Closing a scope cancels its tasks and rejects new work. A task removes
itself when it finishes, including when it fails or is cancelled.
"""
def __init__(self, name: str = "application") -> None:
self.name = name
self.closed = False
self._tasks: set[Any] = set()
@property
def pending(self) -> int:
"""Number of unfinished tasks owned by this scope."""
return sum(not task.done() for task in self._tasks)
def create_task(self, awaitable: Awaitlike[T], *, report_errors: bool = False) -> Any:
"""Schedule work with this scope's lifetime and context."""
if self.closed:
if inspect.iscoroutine(awaitable):
awaitable.close()
raise RuntimeError(f"Task scope {self.name!r} is closed")
token = _scope.set(self)
try:
task = _schedule(awaitable)
finally:
_scope.reset(token)
self._tasks.add(task)
def finished(done: Any) -> None:
self._tasks.discard(done)
if report_errors and not done.cancelled():
error = done.exception()
if error is not None:
from .diagnostics import report_error
if not report_error(error, phase=f"task in {self.name}"):
get_loop().call_exception_handler(
{"message": f"Task in {self.name} failed", "exception": error}
)
task.add_done_callback(finished)
return task
def close(self) -> None:
"""Cancel owned work; calling this more than once is harmless."""
self.closed = True
for task in tuple(self._tasks):
if (
isinstance(task, asyncio.Future)
and task.get_loop().is_running()
and not _on_loop_thread(task.get_loop())
):
task.get_loop().call_soon_threadsafe(task.cancel)
else:
task.cancel()
self._tasks.clear()
_application_scope = TaskScope()
def start() -> asyncio.AbstractEventLoop:
"""Start the application thread once and return its event loop."""
global _loop, _thread, _owner
with _lock:
if _thread is not None and _thread.is_alive():
assert _loop is not None
return _loop
if _loop is not None and not _loop.is_closed():
raise RuntimeError("Start the application runtime before creating headless tasks")
ready = threading.Event()
loop = asyncio.new_event_loop()
_loop = loop
def serve() -> None:
asyncio.set_event_loop(loop)
ready.set()
try:
loop.run_forever()
finally:
loop.run_until_complete(_cancel_tasks(loop))
loop.run_until_complete(loop.shutdown_asyncgens())
loop.run_until_complete(loop.shutdown_default_executor())
loop.close()
_thread = threading.Thread(target=serve, name="PythonNative", daemon=True)
_owner = _thread
_thread.start()
ready.wait()
return loop
def get_loop() -> asyncio.AbstractEventLoop:
"""Return the application loop, or the caller's loop in headless mode."""
global _loop, _owner
if _thread is not None and _thread.is_alive():
assert _loop is not None
return _loop
try:
return asyncio.get_running_loop()
except RuntimeError:
with _lock:
if _loop is None or _loop.is_closed():
_loop = asyncio.new_event_loop()
_owner = threading.current_thread()
return _loop
def _on_loop_thread(loop: asyncio.AbstractEventLoop) -> bool:
try:
return asyncio.get_running_loop() is loop
except RuntimeError:
return loop is _loop and threading.current_thread() is _owner
async def _ensure_coro(awaitable: Awaitlike[T]) -> T:
return await awaitable
def _schedule(awaitable: Awaitlike[T]) -> Any:
loop = get_loop()
if _on_loop_thread(loop):
return asyncio.ensure_future(awaitable, loop=loop)
return asyncio.run_coroutine_threadsafe(_ensure_coro(awaitable), loop)
def run_async(awaitable: Awaitlike[T]) -> Any:
"""Schedule work in the current component scope or the application scope.
Returns an asyncio task on the application thread and a concurrent
future on other threads. Both handles support cancellation.
"""
return (_scope.get() or _application_scope).create_task(awaitable)
def run_application_task(awaitable: Awaitlike[T]) -> Any:
"""Start work that should survive the component that requested it."""
return _application_scope.create_task(awaitable)
def invoke(callback: Callable[..., Any], *args: Any, scope: TaskScope | None = None) -> Any:
"""Invoke a callback and schedule an awaitable result in its owner's scope."""
owner = scope or _scope.get() or _application_scope
if owner.closed:
return None
token = _scope.set(owner)
try:
result = callback(*args)
if inspect.isawaitable(result):
return owner.create_task(result, report_errors=True)
return result
finally:
_scope.reset(token)
def run_blocking(awaitable: Awaitlike[T], timeout: float | None = None) -> T:
"""Wait for work from synchronous code; never block the application loop."""
loop = get_loop()
if loop.is_running() and _on_loop_thread(loop):
if inspect.iscoroutine(awaitable):
awaitable.close()
raise RuntimeError("run_blocking() cannot run inside the event loop; await the result")
async def bounded() -> T:
async with asyncio.timeout(timeout):
return await awaitable
if loop.is_running():
return asyncio.run_coroutine_threadsafe(bounded(), loop).result()
return loop.run_until_complete(bounded())
def drain(timeout: float = 1.0, *, until: Callable[[], bool] | None = None) -> bool:
"""Settle headless work or wait for application tasks from a test thread.
Long-lived application tasks require an explicit ``until`` predicate.
This uses public asyncio APIs and never inspects selector internals.
"""
async def settle() -> bool:
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout
idle_turns = 0
while loop.time() < deadline:
await asyncio.sleep(0)
if until is not None and until():
return True
pending = [task for task in asyncio.all_tasks() if task is not asyncio.current_task() and not task.done()]
if until is None and not pending:
idle_turns += 1
if idle_turns >= 3:
return True
else:
idle_turns = 0
await asyncio.sleep(0.001)
return False
return run_blocking(settle())
def call_threadsafe(callback: Callable[..., Any], *args: Any) -> None:
"""Enqueue a callback on the application event loop from any thread."""
get_loop().call_soon_threadsafe(callback, *args)
def call_on_application_thread(fn: Callable[[], None]) -> None:
"""Execute Python work on its owner thread, inline when already there."""
loop = get_loop()
if _on_loop_thread(loop):
fn()
else:
loop.call_soon_threadsafe(fn)
def resolve_future(future: asyncio.Future[T], value: T) -> None:
"""Complete a future from any thread unless it was cancelled."""
def complete() -> None:
if not future.done():
future.set_result(value)
future.get_loop().call_soon_threadsafe(complete)
def reject_future(future: asyncio.Future[Any], error: BaseException) -> None:
"""Fail a future from any thread unless it was cancelled."""
def complete() -> None:
if not future.done():
future.set_exception(error)
future.get_loop().call_soon_threadsafe(complete)
def create_future() -> asyncio.Future[Any]:
"""Create a future for completion by a native request."""
return get_loop().create_future()
async def _cancel_tasks(loop: asyncio.AbstractEventLoop) -> None:
tasks = [task for task in asyncio.all_tasks(loop) if task is not asyncio.current_task()]
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
def _shutdown_for_tests() -> None:
global _loop, _thread, _owner, _application_scope
with _lock:
loop, thread = _loop, _thread
_application_scope.close()
if loop is not None and not loop.is_closed():
if thread is not None:
loop.call_soon_threadsafe(loop.stop)
thread.join(timeout=5)
if thread.is_alive():
raise RuntimeError("Application thread did not stop")
elif not loop.is_running():
loop.run_until_complete(_cancel_tasks(loop))
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
_loop = _thread = _owner = None
_application_scope = TaskScope()
__all__ = [
"TaskScope",
"call_on_application_thread",
"call_threadsafe",
"create_future",
"drain",
"get_loop",
"reject_future",
"resolve_future",
"run_application_task",
"run_async",
"run_blocking",
"start",
]