Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/from_thread.py: 33%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3__all__ = (
4 "BlockingPortal",
5 "BlockingPortalProvider",
6 "check_cancelled",
7 "run",
8 "run_sync",
9 "start_blocking_portal",
10)
12import sys
13from collections.abc import Awaitable, Callable, Coroutine, Generator
14from concurrent.futures import Future
15from contextlib import (
16 AbstractAsyncContextManager,
17 AbstractContextManager,
18 contextmanager,
19)
20from dataclasses import dataclass, field
21from functools import partial
22from inspect import isawaitable
23from threading import Lock, Thread, current_thread, get_ident
24from types import TracebackType
25from typing import (
26 Any,
27 Generic,
28 TypeVar,
29 cast,
30 overload,
31)
33from ._core._eventloop import (
34 get_cancelled_exc_class,
35 threadlocals,
36)
37from ._core._eventloop import run as run_eventloop
38from ._core._exceptions import NoEventLoopError
39from ._core._synchronization import Event
40from ._core._tasks import CancelScope, create_task_group
41from .abc._tasks import TaskStatus
42from .lowlevel import EventLoopToken, current_token
44if sys.version_info >= (3, 11):
45 from typing import TypeVarTuple, Unpack
46else:
47 from typing_extensions import TypeVarTuple, Unpack
49T_Retval = TypeVar("T_Retval")
50T_co = TypeVar("T_co", covariant=True)
51PosArgsT = TypeVarTuple("PosArgsT")
54def _token_or_error(token: EventLoopToken | None) -> EventLoopToken:
55 if token is not None:
56 return token
58 try:
59 return threadlocals.current_token
60 except AttributeError:
61 raise NoEventLoopError(
62 "Not running inside an AnyIO worker thread, and no event loop token was "
63 "provided"
64 ) from None
67def run(
68 func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]],
69 *args: Unpack[PosArgsT],
70 token: EventLoopToken | None = None,
71) -> T_co:
72 """
73 Call a coroutine function from a worker thread.
75 :param func: a coroutine function
76 :param args: positional arguments for the callable
77 :param token: an event loop token to use to get back to the event loop thread
78 (required if calling this function from outside an AnyIO worker thread)
79 :return: the return value of the coroutine function
80 :raises MissingTokenError: if no token was provided and called from outside an
81 AnyIO worker thread
82 :raises RunFinishedError: if the event loop tied to ``token`` is no longer running
84 .. versionchanged:: 4.11.0
85 Added the ``token`` parameter.
87 """
88 explicit_token = token is not None
89 token = _token_or_error(token)
90 return token.backend_class.run_async_from_thread(
91 func, args, token=token.native_token if explicit_token else None
92 )
95def run_sync(
96 func: Callable[[Unpack[PosArgsT]], T_Retval],
97 *args: Unpack[PosArgsT],
98 token: EventLoopToken | None = None,
99) -> T_Retval:
100 """
101 Call a function in the event loop thread from a worker thread.
103 :param func: a callable
104 :param args: positional arguments for the callable
105 :param token: an event loop token to use to get back to the event loop thread
106 (required if calling this function from outside an AnyIO worker thread)
107 :return: the return value of the callable
108 :raises MissingTokenError: if no token was provided and called from outside an
109 AnyIO worker thread
110 :raises RunFinishedError: if the event loop tied to ``token`` is no longer running
112 .. versionchanged:: 4.11.0
113 Added the ``token`` parameter.
115 """
116 explicit_token = token is not None
117 token = _token_or_error(token)
118 return token.backend_class.run_sync_from_thread(
119 func, args, token=token.native_token if explicit_token else None
120 )
123class _BlockingAsyncContextManager(Generic[T_co], AbstractContextManager):
124 _enter_future: Future[T_co]
125 _exit_future: Future[bool | None]
126 _exit_event: Event
127 _exit_exc_info: tuple[
128 type[BaseException] | None, BaseException | None, TracebackType | None
129 ] = (None, None, None)
131 def __init__(
132 self, async_cm: AbstractAsyncContextManager[T_co], portal: BlockingPortal
133 ):
134 self._async_cm = async_cm
135 self._portal = portal
137 async def run_async_cm(self) -> bool | None:
138 try:
139 self._exit_event = Event()
140 value = await self._async_cm.__aenter__()
141 except BaseException as exc:
142 self._enter_future.set_exception(exc)
143 raise
144 else:
145 self._enter_future.set_result(value)
147 try:
148 # Wait for the sync context manager to exit.
149 # This next statement can raise `get_cancelled_exc_class()` if
150 # something went wrong in a task group in this async context
151 # manager.
152 await self._exit_event.wait()
153 finally:
154 # In case of cancellation, it could be that we end up here before
155 # `_BlockingAsyncContextManager.__exit__` is called, and an
156 # `_exit_exc_info` has been set.
157 result = await self._async_cm.__aexit__(*self._exit_exc_info)
159 return result
161 def __enter__(self) -> T_co:
162 self._enter_future = Future()
163 self._exit_future = self._portal.start_task_soon(self.run_async_cm)
164 return self._enter_future.result()
166 def __exit__(
167 self,
168 __exc_type: type[BaseException] | None,
169 __exc_value: BaseException | None,
170 __traceback: TracebackType | None,
171 ) -> bool | None:
172 self._exit_exc_info = __exc_type, __exc_value, __traceback
173 self._portal.call(self._exit_event.set)
174 return self._exit_future.result()
177class _BlockingPortalTaskStatus(TaskStatus):
178 def __init__(self, future: Future):
179 self._future = future
181 def started(self, value: object = None) -> None:
182 self._future.set_result(value)
185class BlockingPortal:
186 """
187 An object that lets external threads run code in an asynchronous event loop.
189 :raises NoEventLoopError: if no supported asynchronous event loop is running in the
190 current thread
191 """
193 def __init__(self) -> None:
194 self._token = current_token()
195 self._event_loop_thread_id: int | None = get_ident()
196 self._stop_event = Event()
197 self._task_group = create_task_group()
199 async def __aenter__(self) -> BlockingPortal:
200 await self._task_group.__aenter__()
201 return self
203 async def __aexit__(
204 self,
205 exc_type: type[BaseException] | None,
206 exc_val: BaseException | None,
207 exc_tb: TracebackType | None,
208 ) -> bool:
209 await self.stop()
210 return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
212 def _check_running(self) -> None:
213 if self._event_loop_thread_id is None:
214 raise RuntimeError("This portal is not running")
215 if self._event_loop_thread_id == get_ident():
216 raise RuntimeError(
217 "This method cannot be called from the event loop thread"
218 )
220 async def sleep_until_stopped(self) -> None:
221 """Sleep until :meth:`stop` is called."""
222 await self._stop_event.wait()
224 async def stop(self, cancel_remaining: bool = False) -> None:
225 """
226 Signal the portal to shut down.
228 This marks the portal as no longer accepting new calls and exits from
229 :meth:`sleep_until_stopped`.
231 :param cancel_remaining: ``True`` to cancel all the remaining tasks, ``False``
232 to let them finish before returning
234 """
235 self._event_loop_thread_id = None
236 self._stop_event.set()
237 if cancel_remaining:
238 self._task_group.cancel_scope.cancel("the blocking portal is shutting down")
240 async def _call_func(
241 self,
242 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
243 args: tuple[Unpack[PosArgsT]],
244 kwargs: dict[str, Any],
245 future: Future[T_Retval],
246 ) -> None:
247 event_loop_thread_id = self._event_loop_thread_id
249 def callback(f: Future[T_Retval]) -> None:
250 if f.cancelled():
251 if event_loop_thread_id == get_ident():
252 scope.cancel("the future was cancelled")
253 elif event_loop_thread_id is not None:
254 run_sync(
255 scope.cancel, "the future was cancelled", token=self._token
256 )
258 try:
259 retval_or_awaitable = func(*args, **kwargs)
260 if isawaitable(retval_or_awaitable):
261 with CancelScope() as scope:
262 future.add_done_callback(callback)
263 retval = await retval_or_awaitable
264 else:
265 retval = retval_or_awaitable
266 except get_cancelled_exc_class():
267 future.cancel()
268 future.set_running_or_notify_cancel()
269 except BaseException as exc:
270 if not future.cancelled():
271 future.set_exception(exc)
273 # Let base exceptions fall through
274 if not isinstance(exc, Exception):
275 raise
276 else:
277 if not future.cancelled():
278 future.set_result(retval)
279 finally:
280 scope = None # type: ignore[assignment]
282 def _spawn_task_from_thread(
283 self,
284 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
285 args: tuple[Unpack[PosArgsT]],
286 kwargs: dict[str, Any],
287 name: object,
288 future: Future[T_Retval],
289 ) -> None:
290 """
291 Spawn a new task using the given callable.
293 :param func: a callable
294 :param args: positional arguments to be passed to the callable
295 :param kwargs: keyword arguments to be passed to the callable
296 :param name: name of the task (will be coerced to a string if not ``None``)
297 :param future: a future that will resolve to the return value of the callable,
298 or the exception raised during its execution
300 """
301 run_sync(
302 partial(self._task_group.start_soon, name=name),
303 self._call_func,
304 func,
305 args,
306 kwargs,
307 future,
308 token=self._token,
309 )
311 @overload
312 def call(
313 self,
314 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
315 *args: Unpack[PosArgsT],
316 ) -> T_Retval: ...
318 @overload
319 def call(
320 self, func: Callable[[Unpack[PosArgsT]], T_Retval], *args: Unpack[PosArgsT]
321 ) -> T_Retval: ...
323 def call(
324 self,
325 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
326 *args: Unpack[PosArgsT],
327 ) -> T_Retval:
328 """
329 Call the given function in the event loop thread.
331 If the callable returns a coroutine object, it is awaited on.
333 :param func: any callable
334 :raises RuntimeError: if the portal is not running or if this method is called
335 from within the event loop thread
337 """
338 return cast(T_Retval, self.start_task_soon(func, *args).result())
340 @overload
341 def start_task_soon(
342 self,
343 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
344 *args: Unpack[PosArgsT],
345 name: object = None,
346 ) -> Future[T_Retval]: ...
348 @overload
349 def start_task_soon(
350 self,
351 func: Callable[[Unpack[PosArgsT]], T_Retval],
352 *args: Unpack[PosArgsT],
353 name: object = None,
354 ) -> Future[T_Retval]: ...
356 def start_task_soon(
357 self,
358 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
359 *args: Unpack[PosArgsT],
360 name: object = None,
361 ) -> Future[T_Retval]:
362 """
363 Start a task in the portal's task group.
365 The task will be run inside a cancel scope which can be cancelled by cancelling
366 the returned future.
368 :param func: the target function
369 :param args: positional arguments passed to ``func``
370 :param name: name of the task (will be coerced to a string if not ``None``)
371 :return: a future that resolves with the return value of the callable if the
372 task completes successfully, or with the exception raised in the task
373 :raises RuntimeError: if the portal is not running or if this method is called
374 from within the event loop thread
375 :rtype: concurrent.futures.Future[T_Retval]
377 .. versionadded:: 3.0
379 """
380 self._check_running()
381 f: Future[T_Retval] = Future()
382 self._spawn_task_from_thread(func, args, {}, name, f)
383 return f
385 def start_task(
386 self,
387 func: Callable[..., Awaitable[T_Retval]],
388 *args: object,
389 name: object = None,
390 ) -> tuple[Future[T_Retval], Any]:
391 """
392 Start a task in the portal's task group and wait until it signals for readiness.
394 This method works the same way as :meth:`.abc.TaskGroup.start`.
396 :param func: the target function
397 :param args: positional arguments passed to ``func``
398 :param name: name of the task (will be coerced to a string if not ``None``)
399 :return: a tuple of (future, task_status_value) where the ``task_status_value``
400 is the value passed to ``task_status.started()`` from within the target
401 function
402 :rtype: tuple[concurrent.futures.Future[T_Retval], Any]
404 .. versionadded:: 3.0
406 """
408 def task_done(future: Future[T_Retval]) -> None:
409 if not task_status_future.done():
410 if future.cancelled():
411 task_status_future.cancel()
412 elif future.exception():
413 task_status_future.set_exception(future.exception())
414 else:
415 exc = RuntimeError(
416 "Task exited without calling task_status.started()"
417 )
418 task_status_future.set_exception(exc)
420 self._check_running()
421 task_status_future: Future = Future()
422 task_status = _BlockingPortalTaskStatus(task_status_future)
423 f: Future = Future()
424 f.add_done_callback(task_done)
425 self._spawn_task_from_thread(func, args, {"task_status": task_status}, name, f)
426 return f, task_status_future.result()
428 def wrap_async_context_manager(
429 self, cm: AbstractAsyncContextManager[T_co]
430 ) -> AbstractContextManager[T_co]:
431 """
432 Wrap an async context manager as a synchronous context manager via this portal.
434 Spawns a task that will call both ``__aenter__()`` and ``__aexit__()``, stopping
435 in the middle until the synchronous context manager exits.
437 :param cm: an asynchronous context manager
438 :return: a synchronous context manager
440 .. versionadded:: 2.1
442 """
443 return _BlockingAsyncContextManager(cm, self)
446@dataclass
447class BlockingPortalProvider:
448 """
449 A manager for a blocking portal. Used as a context manager. The first thread to
450 enter this context manager causes a blocking portal to be started with the specific
451 parameters, and the last thread to exit causes the portal to be shut down. Thus,
452 there will be exactly one blocking portal running in this context as long as at
453 least one thread has entered this context manager.
455 The parameters are the same as for :func:`~anyio.run`.
457 :param backend: name of the backend
458 :param backend_options: backend options
460 .. versionadded:: 4.4
461 """
463 backend: str = "asyncio"
464 backend_options: dict[str, Any] | None = None
465 _lock: Lock = field(init=False, default_factory=Lock)
466 _leases: int = field(init=False, default=0)
467 _portal: BlockingPortal = field(init=False)
468 _portal_cm: AbstractContextManager[BlockingPortal] | None = field(
469 init=False, default=None
470 )
472 def __enter__(self) -> BlockingPortal:
473 with self._lock:
474 if self._portal_cm is None:
475 self._portal_cm = start_blocking_portal(
476 self.backend, self.backend_options
477 )
478 self._portal = self._portal_cm.__enter__()
480 self._leases += 1
481 return self._portal
483 def __exit__(
484 self,
485 exc_type: type[BaseException] | None,
486 exc_val: BaseException | None,
487 exc_tb: TracebackType | None,
488 ) -> None:
489 portal_cm: AbstractContextManager[BlockingPortal] | None = None
490 with self._lock:
491 assert self._portal_cm
492 assert self._leases > 0
493 self._leases -= 1
494 if not self._leases:
495 portal_cm = self._portal_cm
496 self._portal_cm = None
497 del self._portal
499 if portal_cm:
500 portal_cm.__exit__(None, None, None)
503@contextmanager
504def start_blocking_portal(
505 backend: str = "asyncio",
506 backend_options: dict[str, Any] | None = None,
507 *,
508 name: str | None = None,
509) -> Generator[BlockingPortal, Any, None]:
510 """
511 Start a new event loop in a new thread and run a blocking portal in its main task.
513 The parameters are the same as for :func:`~anyio.run`.
515 :param backend: name of the backend
516 :param backend_options: backend options
517 :param name: name of the thread
518 :return: a context manager that yields a blocking portal
520 .. versionchanged:: 3.0
521 Usage as a context manager is now required.
523 """
525 async def run_portal() -> None:
526 async with BlockingPortal() as portal_:
527 if name is None:
528 current_thread().name = f"{backend}-portal-{id(portal_):x}"
530 future.set_result(portal_)
531 await portal_.sleep_until_stopped()
533 def run_blocking_portal() -> None:
534 if future.set_running_or_notify_cancel():
535 try:
536 run_eventloop(
537 run_portal, backend=backend, backend_options=backend_options
538 )
539 except BaseException as exc:
540 if not future.done():
541 future.set_exception(exc)
543 future: Future[BlockingPortal] = Future()
544 thread = Thread(target=run_blocking_portal, daemon=True, name=name)
545 thread.start()
546 try:
547 cancel_remaining_tasks = False
548 portal = future.result()
549 try:
550 yield portal
551 except BaseException:
552 cancel_remaining_tasks = True
553 raise
554 finally:
555 try:
556 portal.call(portal.stop, cancel_remaining_tasks)
557 except RuntimeError:
558 pass
559 finally:
560 thread.join()
563def check_cancelled() -> None:
564 """
565 Check if the cancel scope of the host task's running the current worker thread has
566 been cancelled.
568 If the host task's current cancel scope has indeed been cancelled, the
569 backend-specific cancellation exception will be raised.
571 :raises RuntimeError: if the current thread was not spawned by
572 :func:`.to_thread.run_sync`
574 """
575 try:
576 token: EventLoopToken = threadlocals.current_token
577 except AttributeError:
578 raise NoEventLoopError(
579 "This function can only be called inside an AnyIO worker thread"
580 ) from None
582 token.backend_class.check_cancelled()