codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import array4import asyncio5import concurrent.futures6import contextvars7import math8import os9import socket10import sys11import threading12import weakref13from asyncio import (14 AbstractEventLoop,15 CancelledError,16 all_tasks,17 create_task,18 current_task,19 get_running_loop,20 sleep,21)22from asyncio.base_events import _run_until_complete_cb # type: ignore[attr-defined]23from collections import OrderedDict, deque24from collections.abc import (25 AsyncGenerator,26 AsyncIterator,27 Awaitable,28 Callable,29 Collection,30 Coroutine,31 Iterable,32 Sequence,33)34from concurrent.futures import Future35from contextlib import AbstractContextManager, suppress36from contextvars import Context, copy_context37from dataclasses import dataclass, field38from functools import partial, wraps39from inspect import (40 CORO_RUNNING,41 CORO_SUSPENDED,42 getcoroutinestate,43 iscoroutine,44)45from io import IOBase46from os import PathLike47from queue import Queue48from signal import Signals49from socket import AddressFamily, SocketKind50from threading import Thread51from types import CodeType, TracebackType52from typing import (53 IO,54 TYPE_CHECKING,55 Any,56 ParamSpec,57 TypeVar,58 cast,59)60from weakref import WeakKeyDictionary61 62from .. import (63 CapacityLimiterStatistics,64 EventStatistics,65 LockStatistics,66 TaskInfo,67 abc,68)69from .._core._eventloop import (70 claim_worker_thread,71 set_current_async_library,72 threadlocals,73)74from .._core._exceptions import (75 BrokenResourceError,76 BusyResourceError,77 ClosedResourceError,78 EndOfStream,79 RunFinishedError,80 WouldBlock,81)82from .._core._sockets import convert_ipv6_sockaddr83from .._core._streams import create_memory_object_stream84from .._core._synchronization import (85 CapacityLimiter as BaseCapacityLimiter,86)87from .._core._synchronization import Event as BaseEvent88from .._core._synchronization import Lock as BaseLock89from .._core._synchronization import (90 ResourceGuard,91 SemaphoreStatistics,92)93from .._core._synchronization import Semaphore as BaseSemaphore94from .._core._tasks import CancelScope as BaseCancelScope95from ..abc import (96 AsyncBackend,97 IPSockAddrType,98 SocketListener,99 UDPPacketType,100 UNIXDatagramPacketType,101)102from ..abc._eventloop import StrOrBytesPath103from ..lowlevel import RunVar104from ..streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream105 106if TYPE_CHECKING:107 from _typeshed import FileDescriptorLike108else:109 FileDescriptorLike = object110 111if sys.version_info >= (3, 11):112 from asyncio import Runner113 from typing import TypeVarTuple, Unpack114else:115 import contextvars116 import enum117 import signal118 from asyncio import coroutines, events, exceptions, tasks119 120 from exceptiongroup import BaseExceptionGroup121 from typing_extensions import TypeVarTuple, Unpack122 123 class _State(enum.Enum):124 CREATED = "created"125 INITIALIZED = "initialized"126 CLOSED = "closed"127 128 class Runner:129 # Copied from CPython 3.11130 def __init__(131 self,132 *,133 debug: bool | None = None,134 loop_factory: Callable[[], AbstractEventLoop] | None = None,135 ):136 self._state = _State.CREATED137 self._debug = debug138 self._loop_factory = loop_factory139 self._loop: AbstractEventLoop | None = None140 self._context = None141 self._interrupt_count = 0142 self._set_event_loop = False143 144 def __enter__(self) -> Runner:145 self._lazy_init()146 return self147 148 def __exit__(149 self,150 exc_type: type[BaseException] | None,151 exc_val: BaseException | None,152 exc_tb: TracebackType | None,153 ) -> None:154 self.close()155 156 def close(self) -> None:157 """Shutdown and close event loop."""158 loop = self._loop159 if self._state is not _State.INITIALIZED or loop is None:160 return161 try:162 _cancel_all_tasks(loop)163 loop.run_until_complete(loop.shutdown_asyncgens())164 if hasattr(loop, "shutdown_default_executor"):165 loop.run_until_complete(loop.shutdown_default_executor())166 else:167 loop.run_until_complete(_shutdown_default_executor(loop))168 finally:169 if self._set_event_loop:170 events.set_event_loop(None)171 loop.close()172 self._loop = None173 self._state = _State.CLOSED174 175 def get_loop(self) -> AbstractEventLoop:176 """Return embedded event loop."""177 self._lazy_init()178 return self._loop179 180 def run(self, coro: Coroutine[T_Retval], *, context=None) -> T_Retval:181 """Run a coroutine inside the embedded event loop."""182 if not coroutines.iscoroutine(coro):183 raise ValueError(f"a coroutine was expected, got {coro!r}")184 185 if events._get_running_loop() is not None:186 # fail fast with short traceback187 raise RuntimeError(188 "Runner.run() cannot be called from a running event loop"189 )190 191 self._lazy_init()192 193 if context is None:194 context = self._context195 task = context.run(self._loop.create_task, coro)196 197 if (198 threading.current_thread() is threading.main_thread()199 and signal.getsignal(signal.SIGINT) is signal.default_int_handler200 ):201 sigint_handler = partial(self._on_sigint, main_task=task)202 try:203 signal.signal(signal.SIGINT, sigint_handler)204 except ValueError:205 # `signal.signal` may throw if `threading.main_thread` does206 # not support signals (e.g. embedded interpreter with signals207 # not registered - see gh-91880)208 sigint_handler = None209 else:210 sigint_handler = None211 212 self._interrupt_count = 0213 try:214 return self._loop.run_until_complete(task)215 except exceptions.CancelledError:216 if self._interrupt_count > 0:217 uncancel = getattr(task, "uncancel", None)218 if uncancel is not None and uncancel() == 0:219 raise KeyboardInterrupt # noqa: B904220 raise # CancelledError221 finally:222 if (223 sigint_handler is not None224 and signal.getsignal(signal.SIGINT) is sigint_handler225 ):226 signal.signal(signal.SIGINT, signal.default_int_handler)227 228 def _lazy_init(self) -> None:229 if self._state is _State.CLOSED:230 raise RuntimeError("Runner is closed")231 if self._state is _State.INITIALIZED:232 return233 if self._loop_factory is None:234 self._loop = events.new_event_loop()235 if not self._set_event_loop:236 # Call set_event_loop only once to avoid calling237 # attach_loop multiple times on child watchers238 events.set_event_loop(self._loop)239 self._set_event_loop = True240 else:241 self._loop = self._loop_factory()242 if self._debug is not None:243 self._loop.set_debug(self._debug)244 self._context = contextvars.copy_context()245 self._state = _State.INITIALIZED246 247 def _on_sigint(self, signum, frame, main_task: asyncio.Task) -> None:248 self._interrupt_count += 1249 if self._interrupt_count == 1 and not main_task.done():250 main_task.cancel()251 # wakeup loop if it is blocked by select() with long timeout252 self._loop.call_soon_threadsafe(lambda: None)253 return254 raise KeyboardInterrupt()255 256 def _cancel_all_tasks(loop: AbstractEventLoop) -> None:257 to_cancel = tasks.all_tasks(loop)258 if not to_cancel:259 return260 261 for task in to_cancel:262 task.cancel()263 264 loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True))265 266 for task in to_cancel:267 if task.cancelled():268 continue269 if task.exception() is not None:270 loop.call_exception_handler(271 {272 "message": "unhandled exception during asyncio.run() shutdown",273 "exception": task.exception(),274 "task": task,275 }276 )277 278 async def _shutdown_default_executor(loop: AbstractEventLoop) -> None:279 """Schedule the shutdown of the default executor."""280 281 def _do_shutdown(future: asyncio.futures.Future) -> None:282 try:283 loop._default_executor.shutdown(wait=True) # type: ignore[attr-defined]284 loop.call_soon_threadsafe(future.set_result, None)285 except Exception as ex:286 loop.call_soon_threadsafe(future.set_exception, ex)287 288 loop._executor_shutdown_called = True289 if loop._default_executor is None:290 return291 future = loop.create_future()292 thread = threading.Thread(target=_do_shutdown, args=(future,))293 thread.start()294 try:295 await future296 finally:297 thread.join()298 299 300T_Retval = TypeVar("T_Retval")301T_contra = TypeVar("T_contra", contravariant=True)302PosArgsT = TypeVarTuple("PosArgsT")303P = ParamSpec("P")304 305_root_task: RunVar[asyncio.Task | None] = RunVar("_root_task")306 307 308def find_root_task() -> asyncio.Task:309 root_task = _root_task.get(None)310 if root_task is not None and not root_task.done():311 return root_task312 313 # Look for a task that has been started via run_until_complete()314 for task in all_tasks():315 if task._callbacks and not task.done():316 callbacks = [cb for cb, context in task._callbacks]317 for cb in callbacks:318 if (319 cb is _run_until_complete_cb320 or getattr(cb, "__module__", None) == "uvloop.loop"321 ):322 _root_task.set(task)323 return task324 325 # Look up the topmost task in the AnyIO task tree, if possible326 task = cast(asyncio.Task, current_task())327 state = _task_states.get(task)328 if state:329 cancel_scope = state.cancel_scope330 while cancel_scope and cancel_scope._parent_scope is not None:331 cancel_scope = cancel_scope._parent_scope332 333 if cancel_scope is not None:334 return cast(asyncio.Task, cancel_scope._host_task)335 336 return task337 338 339def get_callable_name(func: Callable) -> str:340 module = getattr(func, "__module__", None)341 qualname = getattr(func, "__qualname__", None)342 return ".".join([x for x in (module, qualname) if x])343 344 345#346# Event loop347#348 349_run_vars: WeakKeyDictionary[asyncio.AbstractEventLoop, Any] = WeakKeyDictionary()350 351 352def _task_started(task: asyncio.Task) -> bool:353 """Return ``True`` if the task has been started and has not finished."""354 # The task coro should never be None here, as we never add finished tasks to the355 # task list356 coro = task.get_coro()357 assert coro is not None358 try:359 return getcoroutinestate(coro) in (CORO_RUNNING, CORO_SUSPENDED)360 except AttributeError:361 # task coro is async_genenerator_asend https://bugs.python.org/issue37771362 raise Exception(f"Cannot determine if task {task} has started or not") from None363 364 365#366# Timeouts and cancellation367#368 369 370def is_anyio_cancellation(exc: CancelledError) -> bool:371 # Sometimes third party frameworks catch a CancelledError and raise a new one, so as372 # a workaround we have to look at the previous ones in __context__ too for a373 # matching cancel message374 while True:375 if (376 exc.args377 and isinstance(exc.args[0], str)378 and exc.args[0].startswith("Cancelled via cancel scope ")379 ):380 return True381 382 if isinstance(exc.__context__, CancelledError):383 exc = exc.__context__384 continue385 386 return False387 388 389class CancelScope(BaseCancelScope):390 def __new__(391 cls, *, deadline: float = math.inf, shield: bool = False392 ) -> CancelScope:393 return object.__new__(cls)394 395 def __init__(self, deadline: float = math.inf, shield: bool = False):396 self._deadline = deadline397 self._shield = shield398 self._parent_scope: CancelScope | None = None399 self._child_scopes: set[CancelScope] = set()400 self._cancel_called = False401 self._cancel_reason: str | None = None402 self._cancelled_caught = False403 self._active = False404 self._timeout_handle: asyncio.TimerHandle | None = None405 self._cancel_handle: asyncio.Handle | None = None406 self._tasks: set[asyncio.Task] = set()407 self._host_task: asyncio.Task | None = None408 if sys.version_info >= (3, 11):409 self._pending_uncancellations: int | None = 0410 else:411 self._pending_uncancellations = None412 413 def __enter__(self) -> CancelScope:414 if self._active:415 raise RuntimeError(416 "Each CancelScope may only be used for a single 'with' block"417 )418 419 self._host_task = host_task = cast(asyncio.Task, current_task())420 self._tasks.add(host_task)421 try:422 task_state = _task_states[host_task]423 except KeyError:424 task_state = TaskState(None, self)425 _task_states[host_task] = task_state426 else:427 self._parent_scope = task_state.cancel_scope428 task_state.cancel_scope = self429 if self._parent_scope is not None:430 # If using an eager task factory, the parent scope may not even contain431 # the host task432 self._parent_scope._child_scopes.add(self)433 self._parent_scope._tasks.discard(host_task)434 435 self._timeout()436 self._active = True437 438 # Start cancelling the host task if the scope was cancelled before entering439 if self._cancel_called:440 self._deliver_cancellation(self)441 442 return self443 444 def __exit__(445 self,446 exc_type: type[BaseException] | None,447 exc_val: BaseException | None,448 exc_tb: TracebackType | None,449 ) -> bool:450 del exc_tb451 452 if not self._active:453 raise RuntimeError("This cancel scope is not active")454 if current_task() is not self._host_task:455 raise RuntimeError(456 "Attempted to exit cancel scope in a different task than it was "457 "entered in"458 )459 460 assert self._host_task is not None461 host_task_state = _task_states.get(self._host_task)462 if host_task_state is None or host_task_state.cancel_scope is not self:463 raise RuntimeError(464 "Attempted to exit a cancel scope that isn't the current tasks's "465 "current cancel scope"466 )467 468 try:469 self._active = False470 if self._timeout_handle:471 self._timeout_handle.cancel()472 self._timeout_handle = None473 474 self._tasks.remove(self._host_task)475 if self._parent_scope is not None:476 self._parent_scope._child_scopes.remove(self)477 self._parent_scope._tasks.add(self._host_task)478 479 host_task_state.cancel_scope = self._parent_scope480 481 # Restart the cancellation effort in the closest visible, cancelled parent482 # scope if necessary483 self._restart_cancellation_in_parent()484 485 # We only swallow the exception iff it was an AnyIO CancelledError, either486 # directly as exc_val or inside an exception group and there are no cancelled487 # parent cancel scopes visible to us here488 if self._cancel_called and not self._parent_cancellation_is_visible_to_us:489 # For each level-cancel() call made on the host task, call uncancel()490 while self._pending_uncancellations:491 self._host_task.uncancel()492 self._pending_uncancellations -= 1493 494 # Update cancelled_caught and check for exceptions we must not swallow495 if isinstance(exc_val, BaseExceptionGroup):496 cancelleds_caught, remaining = exc_val.split(497 lambda exc: (498 isinstance(exc, CancelledError)499 and is_anyio_cancellation(exc)500 )501 )502 503 if cancelleds_caught is None:504 return False505 506 self._cancelled_caught = True507 508 if remaining is None:509 return True510 511 context = remaining.__context__512 try:513 # Preserve __cause__ and __suppress_context__ by avoiding `raise514 # ... from ...`515 raise remaining516 finally:517 # Preserve __context__518 remaining.__context__ = context519 del context520 else:521 if isinstance(exc_val, CancelledError) and is_anyio_cancellation(522 exc_val523 ):524 self._cancelled_caught = True525 return True526 else:527 return False528 else:529 if self._pending_uncancellations:530 assert self._parent_scope is not None531 assert self._parent_scope._pending_uncancellations is not None532 self._parent_scope._pending_uncancellations += (533 self._pending_uncancellations534 )535 self._pending_uncancellations = 0536 537 return False538 finally:539 self._host_task = None540 del exc_val541 542 @property543 def _effectively_cancelled(self) -> bool:544 cancel_scope: CancelScope | None = self545 while cancel_scope is not None:546 if cancel_scope._cancel_called:547 return True548 549 if cancel_scope.shield:550 return False551 552 cancel_scope = cancel_scope._parent_scope553 554 return False555 556 @property557 def _parent_cancellation_is_visible_to_us(self) -> bool:558 return (559 self._parent_scope is not None560 and not self.shield561 and self._parent_scope._effectively_cancelled562 )563 564 def _timeout(self) -> None:565 if self._deadline != math.inf:566 loop = get_running_loop()567 if loop.time() >= self._deadline:568 self.cancel("deadline exceeded")569 else:570 self._timeout_handle = loop.call_at(self._deadline, self._timeout)571 572 def _deliver_cancellation(self, origin: CancelScope) -> bool:573 """574 Deliver cancellation to directly contained tasks and nested cancel scopes.575 576 Schedule another run at the end if we still have tasks eligible for577 cancellation.578 579 :param origin: the cancel scope that originated the cancellation580 :return: ``True`` if the delivery needs to be retried on the next cycle581 582 """583 should_retry = False584 current = current_task()585 for task in self._tasks:586 should_retry = True587 if task._must_cancel: # type: ignore[attr-defined]588 continue589 590 # The task is eligible for cancellation if it has started591 if task is not current and (task is self._host_task or _task_started(task)):592 waiter = task._fut_waiter # type: ignore[attr-defined]593 if not isinstance(waiter, asyncio.Future) or not waiter.done():594 task.cancel(origin._cancel_reason)595 if (596 task is origin._host_task597 and origin._pending_uncancellations is not None598 ):599 origin._pending_uncancellations += 1600 601 # Deliver cancellation to child scopes that aren't shielded or running their own602 # cancellation callbacks603 for scope in self._child_scopes:604 if not scope._shield and not scope.cancel_called:605 should_retry = scope._deliver_cancellation(origin) or should_retry606 607 # Schedule another callback if there are still tasks left608 if origin is self:609 if should_retry:610 self._cancel_handle = get_running_loop().call_soon(611 self._deliver_cancellation, origin612 )613 else:614 self._cancel_handle = None615 616 return should_retry617 618 def _restart_cancellation_in_parent(self) -> None:619 """620 Restart the cancellation effort in the closest directly cancelled parent scope.621 622 """623 scope = self._parent_scope624 while scope is not None:625 if scope._cancel_called:626 if scope._cancel_handle is None:627 scope._deliver_cancellation(scope)628 629 break630 631 # No point in looking beyond any shielded scope632 if scope._shield:633 break634 635 scope = scope._parent_scope636 637 def cancel(self, reason: str | None = None) -> None:638 if not self._cancel_called:639 if self._timeout_handle:640 self._timeout_handle.cancel()641 self._timeout_handle = None642 643 self._cancel_called = True644 self._cancel_reason = f"Cancelled via cancel scope {id(self):x}"645 if task := current_task():646 self._cancel_reason += f" by {task}"647 648 if reason:649 self._cancel_reason += f"; reason: {reason}"650 651 if self._host_task is not None:652 self._deliver_cancellation(self)653 654 @property655 def deadline(self) -> float:656 return self._deadline657 658 @deadline.setter659 def deadline(self, value: float) -> None:660 self._deadline = float(value)661 if self._timeout_handle is not None:662 self._timeout_handle.cancel()663 self._timeout_handle = None664 665 if self._active and not self._cancel_called:666 self._timeout()667 668 @property669 def cancel_called(self) -> bool:670 return self._cancel_called671 672 @property673 def cancelled_caught(self) -> bool:674 return self._cancelled_caught675 676 @property677 def shield(self) -> bool:678 return self._shield679 680 @shield.setter681 def shield(self, value: bool) -> None:682 if self._shield != value:683 self._shield = value684 if not value:685 self._restart_cancellation_in_parent()686 687 688#689# Task states690#691 692 693class TaskState:694 """695 Encapsulates auxiliary task information that cannot be added to the Task instance696 itself because there are no guarantees about its implementation.697 """698 699 __slots__ = "parent_id", "cancel_scope", "__weakref__"700 701 def __init__(self, parent_id: int | None, cancel_scope: CancelScope | None):702 self.parent_id = parent_id703 self.cancel_scope = cancel_scope704 705 706_task_states: WeakKeyDictionary[asyncio.Task, TaskState] = WeakKeyDictionary()707 708 709#710# Task groups711#712 713 714class _AsyncioTaskStatus(abc.TaskStatus):715 def __init__(self, future: asyncio.Future, parent_id: int):716 self._future = future717 self._parent_id = parent_id718 719 def started(self, value: T_contra | None = None) -> None:720 try:721 self._future.set_result(value)722 except asyncio.InvalidStateError:723 if not self._future.cancelled():724 raise RuntimeError(725 "called 'started' twice on the same task status"726 ) from None727 728 task = cast(asyncio.Task, current_task())729 _task_states[task].parent_id = self._parent_id730 731 732if sys.version_info >= (3, 12):733 _eager_task_factory_code: CodeType | None = asyncio.eager_task_factory.__code__734else:735 _eager_task_factory_code = None736 737 738class TaskGroup(abc.TaskGroup):739 def __init__(self) -> None:740 self.cancel_scope: CancelScope = CancelScope()741 self._active = False742 self._exceptions: list[BaseException] = []743 self._tasks: set[asyncio.Task] = set()744 self._on_completed_fut: asyncio.Future[None] | None = None745 746 async def __aenter__(self) -> TaskGroup:747 self.cancel_scope.__enter__()748 self._active = True749 return self750 751 async def __aexit__(752 self,753 exc_type: type[BaseException] | None,754 exc_val: BaseException | None,755 exc_tb: TracebackType | None,756 ) -> bool:757 try:758 if exc_val is not None:759 self.cancel_scope.cancel()760 if not isinstance(exc_val, CancelledError):761 self._exceptions.append(exc_val)762 763 loop = get_running_loop()764 try:765 if self._tasks:766 with CancelScope() as wait_scope:767 while self._tasks:768 self._on_completed_fut = loop.create_future()769 770 try:771 await self._on_completed_fut772 except CancelledError as exc:773 # Shield the scope against further cancellation attempts,774 # as they're not productive (#695)775 wait_scope.shield = True776 self.cancel_scope.cancel()777 778 # Set exc_val from the cancellation exception if it was779 # previously unset. However, we should not replace a native780 # cancellation exception with one raise by a cancel scope.781 if exc_val is None or (782 isinstance(exc_val, CancelledError)783 and not is_anyio_cancellation(exc)784 ):785 exc_val = exc786 787 self._on_completed_fut = None788 else:789 # If there are no child tasks to wait on, run at least one checkpoint790 # anyway791 await AsyncIOBackend.cancel_shielded_checkpoint()792 793 self._active = False794 if self._exceptions:795 # The exception that got us here should already have been796 # added to self._exceptions so it's ok to break exception797 # chaining and avoid adding a "During handling of above..."798 # for each nesting level.799 raise BaseExceptionGroup(800 "unhandled errors in a TaskGroup", self._exceptions801 ) from None802 elif exc_val:803 raise exc_val804 except BaseException as exc:805 if self.cancel_scope.__exit__(type(exc), exc, exc.__traceback__):806 return True807 808 raise809 810 return self.cancel_scope.__exit__(exc_type, exc_val, exc_tb)811 finally:812 del exc_val, exc_tb, self._exceptions813 814 def _spawn(815 self,816 func: Callable[[Unpack[PosArgsT]], Awaitable[Any]],817 args: tuple[Unpack[PosArgsT]],818 name: object,819 task_status_future: asyncio.Future | None = None,820 ) -> asyncio.Task:821 def task_done(_task: asyncio.Task) -> None:822 if sys.version_info >= (3, 14) and self.cancel_scope._host_task is not None:823 asyncio.future_discard_from_awaited_by(824 _task, self.cancel_scope._host_task825 )826 827 task_state = _task_states[_task]828 assert task_state.cancel_scope is not None829 assert _task in task_state.cancel_scope._tasks830 task_state.cancel_scope._tasks.remove(_task)831 self._tasks.remove(task)832 del _task_states[_task]833 834 if self._on_completed_fut is not None and not self._tasks:835 try:836 self._on_completed_fut.set_result(None)837 except asyncio.InvalidStateError:838 pass839 840 try:841 exc = _task.exception()842 except CancelledError as e:843 while isinstance(e.__context__, CancelledError):844 e = e.__context__845 846 exc = e847 848 if exc is not None:849 # The future can only be in the cancelled state if the host task was850 # cancelled, so return immediately instead of adding one more851 # CancelledError to the exceptions list852 if task_status_future is not None and task_status_future.cancelled():853 return854 855 if task_status_future is None or task_status_future.done():856 if not isinstance(exc, CancelledError):857 self._exceptions.append(exc)858 859 if not self.cancel_scope._effectively_cancelled:860 self.cancel_scope.cancel()861 else:862 task_status_future.set_exception(exc)863 elif task_status_future is not None and not task_status_future.done():864 task_status_future.set_exception(865 RuntimeError("Child exited without calling task_status.started()")866 )867 868 if not self._active:869 raise RuntimeError(870 "This task group is not active; no new tasks can be started."871 )872 873 kwargs = {}874 if task_status_future:875 parent_id = id(current_task())876 kwargs["task_status"] = _AsyncioTaskStatus(877 task_status_future, id(self.cancel_scope._host_task)878 )879 else:880 parent_id = id(self.cancel_scope._host_task)881 882 coro = func(*args, **kwargs)883 if not iscoroutine(coro):884 prefix = f"{func.__module__}." if hasattr(func, "__module__") else ""885 raise TypeError(886 f"Expected {prefix}{func.__qualname__}() to return a coroutine, but "887 f"the return value ({coro!r}) is not a coroutine object"888 )889 890 name = get_callable_name(func) if name is None else str(name)891 loop = asyncio.get_running_loop()892 if (893 (factory := loop.get_task_factory())894 and getattr(factory, "__code__", None) is _eager_task_factory_code895 and (closure := getattr(factory, "__closure__", None))896 ):897 custom_task_constructor = closure[0].cell_contents898 task = custom_task_constructor(coro, loop=loop, name=name)899 else:900 task = create_task(coro, name=name)901 902 # Make the spawned task inherit the task group's cancel scope903 _task_states[task] = TaskState(904 parent_id=parent_id, cancel_scope=self.cancel_scope905 )906 self.cancel_scope._tasks.add(task)907 self._tasks.add(task)908 if sys.version_info >= (3, 14) and self.cancel_scope._host_task is not None:909 asyncio.future_add_to_awaited_by(task, self.cancel_scope._host_task)910 911 task.add_done_callback(task_done)912 return task913 914 def start_soon(915 self,916 func: Callable[[Unpack[PosArgsT]], Awaitable[Any]],917 *args: Unpack[PosArgsT],918 name: object = None,919 ) -> None:920 self._spawn(func, args, name)921 922 async def start(923 self, func: Callable[..., Awaitable[Any]], *args: object, name: object = None924 ) -> Any:925 future: asyncio.Future = asyncio.Future()926 task = self._spawn(func, args, name, future)927 928 # If the task raises an exception after sending a start value without a switch929 # point between, the task group is cancelled and this method never proceeds to930 # process the completed future. That's why we have to have a shielded cancel931 # scope here.932 try:933 return await future934 except CancelledError:935 # Cancel the task and wait for it to exit before returning936 task.cancel()937 with CancelScope(shield=True), suppress(CancelledError):938 await task939 940 raise941 942 943#944# Threads945#946 947_Retval_Queue_Type = tuple[T_Retval | None, BaseException | None]948 949 950class WorkerThread(Thread):951 MAX_IDLE_TIME = 10 # seconds952 953 def __init__(954 self,955 root_task: asyncio.Task,956 workers: set[WorkerThread],957 idle_workers: deque[WorkerThread],958 ):959 super().__init__(name="AnyIO worker thread")960 self.root_task = root_task961 self.workers = workers962 self.idle_workers = idle_workers963 self.loop = root_task._loop964 self.queue: Queue[965 tuple[Context, Callable, tuple, asyncio.Future, CancelScope] | None966 ] = Queue(2)967 self.idle_since = AsyncIOBackend.current_time()968 self.stopping = False969 970 def _report_result(971 self, future: asyncio.Future, result: Any, exc: BaseException | None972 ) -> None:973 self.idle_since = AsyncIOBackend.current_time()974 if not self.stopping:975 self.idle_workers.append(self)976 977 if not future.cancelled():978 if exc is not None:979 if isinstance(exc, StopIteration):980 new_exc = RuntimeError("coroutine raised StopIteration")981 new_exc.__cause__ = exc982 exc = new_exc983 984 future.set_exception(exc)985 else:986 future.set_result(result)987 988 def run(self) -> None:989 with claim_worker_thread(AsyncIOBackend, self.loop):990 while True:991 item = self.queue.get()992 if item is None:993 # Shutdown command received994 return995 996 context, func, args, future, cancel_scope = item997 if not future.cancelled():998 result = None999 exception: BaseException | None = None1000 threadlocals.current_cancel_scope = cancel_scope1001 try:1002 result = context.run(func, *args)1003 except BaseException as exc:1004 exception = exc1005 finally:1006 del threadlocals.current_cancel_scope1007 1008 if not self.loop.is_closed():1009 self.loop.call_soon_threadsafe(1010 self._report_result, future, result, exception1011 )1012 1013 del result, exception1014 1015 self.queue.task_done()1016 del item, context, func, args, future, cancel_scope1017 1018 def stop(self, f: asyncio.Task | None = None) -> None:1019 self.stopping = True1020 self.queue.put_nowait(None)1021 self.workers.discard(self)1022 try:1023 self.idle_workers.remove(self)1024 except ValueError:1025 pass1026 1027 1028_threadpool_idle_workers: RunVar[deque[WorkerThread]] = RunVar(1029 "_threadpool_idle_workers"1030)1031_threadpool_workers: RunVar[set[WorkerThread]] = RunVar("_threadpool_workers")1032 1033 1034#1035# Subprocesses1036#1037 1038 1039@dataclass(eq=False)1040class StreamReaderWrapper(abc.ByteReceiveStream):1041 _stream: asyncio.StreamReader1042 1043 async def receive(self, max_bytes: int = 65536) -> bytes:1044 data = await self._stream.read(max_bytes)1045 if data:1046 return data1047 else:1048 raise EndOfStream1049 1050 async def aclose(self) -> None:1051 self._stream.set_exception(ClosedResourceError())1052 await AsyncIOBackend.checkpoint()1053 1054 1055@dataclass(eq=False)1056class StreamWriterWrapper(abc.ByteSendStream):1057 _stream: asyncio.StreamWriter1058 _closed: bool = field(init=False, default=False)1059 1060 async def send(self, item: bytes) -> None:1061 await AsyncIOBackend.checkpoint_if_cancelled()1062 stream_paused = self._stream._protocol._paused # type: ignore[attr-defined]1063 try:1064 self._stream.write(item)1065 await self._stream.drain()1066 except (ConnectionResetError, BrokenPipeError, RuntimeError) as exc:1067 # If closed by us and/or the peer:1068 # * on stdlib, drain() raises ConnectionResetError or BrokenPipeError1069 # * on uvloop and Winloop, write() eventually starts raising RuntimeError1070 if self._closed:1071 raise ClosedResourceError from exc1072 elif self._stream.is_closing():1073 raise BrokenResourceError from exc1074 1075 raise1076 1077 if not stream_paused:1078 await AsyncIOBackend.cancel_shielded_checkpoint()1079 1080 async def aclose(self) -> None:1081 self._closed = True1082 self._stream.close()1083 await AsyncIOBackend.checkpoint()1084 1085 1086@dataclass(eq=False)1087class Process(abc.Process):1088 _process: asyncio.subprocess.Process1089 _stdin: StreamWriterWrapper | None1090 _stdout: StreamReaderWrapper | None1091 _stderr: StreamReaderWrapper | None1092 1093 async def aclose(self) -> None:1094 with CancelScope(shield=True) as scope:1095 if self._stdin:1096 await self._stdin.aclose()1097 if self._stdout:1098 await self._stdout.aclose()1099 if self._stderr:1100 await self._stderr.aclose()1101 1102 scope.shield = False1103 try:1104 await self.wait()1105 except BaseException:1106 scope.shield = True1107 self.kill()1108 await self.wait()1109 raise1110 1111 async def wait(self) -> int:1112 return await self._process.wait()1113 1114 def terminate(self) -> None:1115 self._process.terminate()1116 1117 def kill(self) -> None:1118 self._process.kill()1119 1120 def send_signal(self, signal: int) -> None:1121 self._process.send_signal(signal)1122 1123 @property1124 def pid(self) -> int:1125 return self._process.pid1126 1127 @property1128 def returncode(self) -> int | None:1129 return self._process.returncode1130 1131 @property1132 def stdin(self) -> abc.ByteSendStream | None:1133 return self._stdin1134 1135 @property1136 def stdout(self) -> abc.ByteReceiveStream | None:1137 return self._stdout1138 1139 @property1140 def stderr(self) -> abc.ByteReceiveStream | None:1141 return self._stderr1142 1143 1144def _forcibly_shutdown_process_pool_on_exit(1145 workers: set[Process], _task: object1146) -> None:1147 """1148 Forcibly shuts down worker processes belonging to this event loop."""1149 child_watcher: asyncio.AbstractChildWatcher | None = None # type: ignore[name-defined]1150 if sys.version_info < (3, 12):1151 try:1152 child_watcher = asyncio.get_event_loop_policy().get_child_watcher()1153 except NotImplementedError:1154 pass1155 1156 # Close as much as possible (w/o async/await) to avoid warnings1157 for process in workers.copy():1158 if process.returncode is not None:1159 continue1160 1161 process._stdin._stream._transport.close() # type: ignore[union-attr]1162 process._stdout._stream._transport.close() # type: ignore[union-attr]1163 process._stderr._stream._transport.close() # type: ignore[union-attr]1164 process.kill()1165 if child_watcher:1166 child_watcher.remove_child_handler(process.pid)1167 1168 1169async def _shutdown_process_pool_on_exit(workers: set[abc.Process]) -> None:1170 """1171 Shuts down worker processes belonging to this event loop.1172 1173 NOTE: this only works when the event loop was started using asyncio.run() or1174 anyio.run().1175 1176 """1177 process: abc.Process1178 try:1179 await sleep(math.inf)1180 except asyncio.CancelledError:1181 workers = workers.copy()1182 for process in workers:1183 if process.returncode is None:1184 process.kill()1185 1186 for process in workers:1187 await process.aclose()1188 1189 1190#1191# Sockets and networking1192#1193 1194 1195class StreamProtocol(asyncio.Protocol):1196 read_queue: deque[bytes]1197 read_event: asyncio.Event1198 write_event: asyncio.Event1199 exception: Exception | None = None1200 is_at_eof: bool = False