1  from __future__ import annotations
   2  
   3  import array
   4  import asyncio
   5  import concurrent.futures
   6  import contextvars
   7  import math
   8  import os
   9  import socket
  10  import sys
  11  import threading
  12  import weakref
  13  from asyncio import (
  14      AbstractEventLoop,
  15      CancelledError,
  16      all_tasks,
  17      create_task,
  18      current_task,
  19      get_running_loop,
  20      sleep,
  21  )
  22  from asyncio.base_events import _run_until_complete_cb  # type: ignore[attr-defined]
  23  from collections import OrderedDict, deque
  24  from collections.abc import (
  25      AsyncGenerator,
  26      AsyncIterator,
  27      Awaitable,
  28      Callable,
  29      Collection,
  30      Coroutine,
  31      Iterable,
  32      Sequence,
  33  )
  34  from concurrent.futures import Future
  35  from contextlib import AbstractContextManager, suppress
  36  from contextvars import Context, copy_context
  37  from dataclasses import dataclass
  38  from functools import partial, wraps
  39  from inspect import (
  40      CORO_RUNNING,
  41      CORO_SUSPENDED,
  42      getcoroutinestate,
  43      iscoroutine,
  44  )
  45  from io import IOBase
  46  from os import PathLike
  47  from queue import Queue
  48  from signal import Signals
  49  from socket import AddressFamily, SocketKind
  50  from threading import Thread
  51  from types import CodeType, TracebackType
  52  from typing import (
  53      IO,
  54      TYPE_CHECKING,
  55      Any,
  56      Optional,
  57      TypeVar,
  58      cast,
  59  )
  60  from weakref import WeakKeyDictionary
  61  
  62  import sniffio
  63  
  64  from .. import (
  65      CapacityLimiterStatistics,
  66      EventStatistics,
  67      LockStatistics,
  68      TaskInfo,
  69      abc,
  70  )
  71  from .._core._eventloop import claim_worker_thread, threadlocals
  72  from .._core._exceptions import (
  73      BrokenResourceError,
  74      BusyResourceError,
  75      ClosedResourceError,
  76      EndOfStream,
  77      WouldBlock,
  78      iterate_exceptions,
  79  )
  80  from .._core._sockets import convert_ipv6_sockaddr
  81  from .._core._streams import create_memory_object_stream
  82  from .._core._synchronization import (
  83      CapacityLimiter as BaseCapacityLimiter,
  84  )
  85  from .._core._synchronization import Event as BaseEvent
  86  from .._core._synchronization import Lock as BaseLock
  87  from .._core._synchronization import (
  88      ResourceGuard,
  89      SemaphoreStatistics,
  90  )
  91  from .._core._synchronization import Semaphore as BaseSemaphore
  92  from .._core._tasks import CancelScope as BaseCancelScope
  93  from ..abc import (
  94      AsyncBackend,
  95      IPSockAddrType,
  96      SocketListener,
  97      UDPPacketType,
  98      UNIXDatagramPacketType,
  99  )
 100  from ..abc._eventloop import StrOrBytesPath
 101  from ..lowlevel import RunVar
 102  from ..streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 103  
 104  if TYPE_CHECKING:
 105      from _typeshed import FileDescriptorLike
 106  else:
 107      FileDescriptorLike = object
 108  
 109  if sys.version_info >= (3, 10):
 110      from typing import ParamSpec
 111  else:
 112      from typing_extensions import ParamSpec
 113  
 114  if sys.version_info >= (3, 11):
 115      from asyncio import Runner
 116      from typing import TypeVarTuple, Unpack
 117  else:
 118      import contextvars
 119      import enum
 120      import signal
 121      from asyncio import coroutines, events, exceptions, tasks
 122  
 123      from exceptiongroup import BaseExceptionGroup
 124      from typing_extensions import TypeVarTuple, Unpack
 125  
 126      class _State(enum.Enum):
 127          CREATED = "created"
 128          INITIALIZED = "initialized"
 129          CLOSED = "closed"
 130  
 131      class Runner:
 132          # Copied from CPython 3.11
 133          def __init__(
 134              self,
 135              *,
 136              debug: bool | None = None,
 137              loop_factory: Callable[[], AbstractEventLoop] | None = None,
 138          ):
 139              self._state = _State.CREATED
 140              self._debug = debug
 141              self._loop_factory = loop_factory
 142              self._loop: AbstractEventLoop | None = None
 143              self._context = None
 144              self._interrupt_count = 0
 145              self._set_event_loop = False
 146  
 147          def __enter__(self) -> Runner:
 148              self._lazy_init()
 149              return self
 150  
 151          def __exit__(
 152              self,
 153              exc_type: type[BaseException],
 154              exc_val: BaseException,
 155              exc_tb: TracebackType,
 156          ) -> None:
 157              self.close()
 158  
 159          def close(self) -> None:
 160              """Shutdown and close event loop."""
 161              if self._state is not _State.INITIALIZED:
 162                  return
 163              try:
 164                  loop = self._loop
 165                  _cancel_all_tasks(loop)
 166                  loop.run_until_complete(loop.shutdown_asyncgens())
 167                  if hasattr(loop, "shutdown_default_executor"):
 168                      loop.run_until_complete(loop.shutdown_default_executor())
 169                  else:
 170                      loop.run_until_complete(_shutdown_default_executor(loop))
 171              finally:
 172                  if self._set_event_loop:
 173                      events.set_event_loop(None)
 174                  loop.close()
 175                  self._loop = None
 176                  self._state = _State.CLOSED
 177  
 178          def get_loop(self) -> AbstractEventLoop:
 179              """Return embedded event loop."""
 180              self._lazy_init()
 181              return self._loop
 182  
 183          def run(self, coro: Coroutine[T_Retval], *, context=None) -> T_Retval:
 184              """Run a coroutine inside the embedded event loop."""
 185              if not coroutines.iscoroutine(coro):
 186                  raise ValueError(f"a coroutine was expected, got {coro!r}")
 187  
 188              if events._get_running_loop() is not None:
 189                  # fail fast with short traceback
 190                  raise RuntimeError(
 191                      "Runner.run() cannot be called from a running event loop"
 192                  )
 193  
 194              self._lazy_init()
 195  
 196              if context is None:
 197                  context = self._context
 198              task = context.run(self._loop.create_task, coro)
 199  
 200              if (
 201                  threading.current_thread() is threading.main_thread()
 202                  and signal.getsignal(signal.SIGINT) is signal.default_int_handler
 203              ):
 204                  sigint_handler = partial(self._on_sigint, main_task=task)
 205                  try:
 206                      signal.signal(signal.SIGINT, sigint_handler)
 207                  except ValueError:
 208                      # `signal.signal` may throw if `threading.main_thread` does
 209                      # not support signals (e.g. embedded interpreter with signals
 210                      # not registered - see gh-91880)
 211                      sigint_handler = None
 212              else:
 213                  sigint_handler = None
 214  
 215              self._interrupt_count = 0
 216              try:
 217                  return self._loop.run_until_complete(task)
 218              except exceptions.CancelledError:
 219                  if self._interrupt_count > 0:
 220                      uncancel = getattr(task, "uncancel", None)
 221                      if uncancel is not None and uncancel() == 0:
 222                          raise KeyboardInterrupt()
 223                  raise  # CancelledError
 224              finally:
 225                  if (
 226                      sigint_handler is not None
 227                      and signal.getsignal(signal.SIGINT) is sigint_handler
 228                  ):
 229                      signal.signal(signal.SIGINT, signal.default_int_handler)
 230  
 231          def _lazy_init(self) -> None:
 232              if self._state is _State.CLOSED:
 233                  raise RuntimeError("Runner is closed")
 234              if self._state is _State.INITIALIZED:
 235                  return
 236              if self._loop_factory is None:
 237                  self._loop = events.new_event_loop()
 238                  if not self._set_event_loop:
 239                      # Call set_event_loop only once to avoid calling
 240                      # attach_loop multiple times on child watchers
 241                      events.set_event_loop(self._loop)
 242                      self._set_event_loop = True
 243              else:
 244                  self._loop = self._loop_factory()
 245              if self._debug is not None:
 246                  self._loop.set_debug(self._debug)
 247              self._context = contextvars.copy_context()
 248              self._state = _State.INITIALIZED
 249  
 250          def _on_sigint(self, signum, frame, main_task: asyncio.Task) -> None:
 251              self._interrupt_count += 1
 252              if self._interrupt_count == 1 and not main_task.done():
 253                  main_task.cancel()
 254                  # wakeup loop if it is blocked by select() with long timeout
 255                  self._loop.call_soon_threadsafe(lambda: None)
 256                  return
 257              raise KeyboardInterrupt()
 258  
 259      def _cancel_all_tasks(loop: AbstractEventLoop) -> None:
 260          to_cancel = tasks.all_tasks(loop)
 261          if not to_cancel:
 262              return
 263  
 264          for task in to_cancel:
 265              task.cancel()
 266  
 267          loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True))
 268  
 269          for task in to_cancel:
 270              if task.cancelled():
 271                  continue
 272              if task.exception() is not None:
 273                  loop.call_exception_handler(
 274                      {
 275                          "message": "unhandled exception during asyncio.run() shutdown",
 276                          "exception": task.exception(),
 277                          "task": task,
 278                      }
 279                  )
 280  
 281      async def _shutdown_default_executor(loop: AbstractEventLoop) -> None:
 282          """Schedule the shutdown of the default executor."""
 283  
 284          def _do_shutdown(future: asyncio.futures.Future) -> None:
 285              try:
 286                  loop._default_executor.shutdown(wait=True)  # type: ignore[attr-defined]
 287                  loop.call_soon_threadsafe(future.set_result, None)
 288              except Exception as ex:
 289                  loop.call_soon_threadsafe(future.set_exception, ex)
 290  
 291          loop._executor_shutdown_called = True
 292          if loop._default_executor is None:
 293              return
 294          future = loop.create_future()
 295          thread = threading.Thread(target=_do_shutdown, args=(future,))
 296          thread.start()
 297          try:
 298              await future
 299          finally:
 300              thread.join()
 301  
 302  
 303  T_Retval = TypeVar("T_Retval")
 304  T_contra = TypeVar("T_contra", contravariant=True)
 305  PosArgsT = TypeVarTuple("PosArgsT")
 306  P = ParamSpec("P")
 307  
 308  _root_task: RunVar[asyncio.Task | None] = RunVar("_root_task")
 309  
 310  
 311  def find_root_task() -> asyncio.Task:
 312      root_task = _root_task.get(None)
 313      if root_task is not None and not root_task.done():
 314          return root_task
 315  
 316      # Look for a task that has been started via run_until_complete()
 317      for task in all_tasks():
 318          if task._callbacks and not task.done():
 319              callbacks = [cb for cb, context in task._callbacks]
 320              for cb in callbacks:
 321                  if (
 322                      cb is _run_until_complete_cb
 323                      or getattr(cb, "__module__", None) == "uvloop.loop"
 324                  ):
 325                      _root_task.set(task)
 326                      return task
 327  
 328      # Look up the topmost task in the AnyIO task tree, if possible
 329      task = cast(asyncio.Task, current_task())
 330      state = _task_states.get(task)
 331      if state:
 332          cancel_scope = state.cancel_scope
 333          while cancel_scope and cancel_scope._parent_scope is not None:
 334              cancel_scope = cancel_scope._parent_scope
 335  
 336          if cancel_scope is not None:
 337              return cast(asyncio.Task, cancel_scope._host_task)
 338  
 339      return task
 340  
 341  
 342  def get_callable_name(func: Callable) -> str:
 343      module = getattr(func, "__module__", None)
 344      qualname = getattr(func, "__qualname__", None)
 345      return ".".join([x for x in (module, qualname) if x])
 346  
 347  
 348  #
 349  # Event loop
 350  #
 351  
 352  _run_vars: WeakKeyDictionary[asyncio.AbstractEventLoop, Any] = WeakKeyDictionary()
 353  
 354  
 355  def _task_started(task: asyncio.Task) -> bool:
 356      """Return ``True`` if the task has been started and has not finished."""
 357      # The task coro should never be None here, as we never add finished tasks to the
 358      # task list
 359      coro = task.get_coro()
 360      assert coro is not None
 361      try:
 362          return getcoroutinestate(coro) in (CORO_RUNNING, CORO_SUSPENDED)
 363      except AttributeError:
 364          # task coro is async_genenerator_asend https://bugs.python.org/issue37771
 365          raise Exception(f"Cannot determine if task {task} has started or not") from None
 366  
 367  
 368  #
 369  # Timeouts and cancellation
 370  #
 371  
 372  
 373  def is_anyio_cancellation(exc: CancelledError) -> bool:
 374      # Sometimes third party frameworks catch a CancelledError and raise a new one, so as
 375      # a workaround we have to look at the previous ones in __context__ too for a
 376      # matching cancel message
 377      while True:
 378          if (
 379              exc.args
 380              and isinstance(exc.args[0], str)
 381              and exc.args[0].startswith("Cancelled by cancel scope ")
 382          ):
 383              return True
 384  
 385          if isinstance(exc.__context__, CancelledError):
 386              exc = exc.__context__
 387              continue
 388  
 389          return False
 390  
 391  
 392  class CancelScope(BaseCancelScope):
 393      def __new__(
 394          cls, *, deadline: float = math.inf, shield: bool = False
 395      ) -> CancelScope:
 396          return object.__new__(cls)
 397  
 398      def __init__(self, deadline: float = math.inf, shield: bool = False):
 399          self._deadline = deadline
 400          self._shield = shield
 401          self._parent_scope: CancelScope | None = None
 402          self._child_scopes: set[CancelScope] = set()
 403          self._cancel_called = False
 404          self._cancelled_caught = False
 405          self._active = False
 406          self._timeout_handle: asyncio.TimerHandle | None = None
 407          self._cancel_handle: asyncio.Handle | None = None
 408          self._tasks: set[asyncio.Task] = set()
 409          self._host_task: asyncio.Task | None = None
 410          if sys.version_info >= (3, 11):
 411              self._pending_uncancellations: int | None = 0
 412          else:
 413              self._pending_uncancellations = None
 414  
 415      def __enter__(self) -> CancelScope:
 416          if self._active:
 417              raise RuntimeError(
 418                  "Each CancelScope may only be used for a single 'with' block"
 419              )
 420  
 421          self._host_task = host_task = cast(asyncio.Task, current_task())
 422          self._tasks.add(host_task)
 423          try:
 424              task_state = _task_states[host_task]
 425          except KeyError:
 426              task_state = TaskState(None, self)
 427              _task_states[host_task] = task_state
 428          else:
 429              self._parent_scope = task_state.cancel_scope
 430              task_state.cancel_scope = self
 431              if self._parent_scope is not None:
 432                  # If using an eager task factory, the parent scope may not even contain
 433                  # the host task
 434                  self._parent_scope._child_scopes.add(self)
 435                  self._parent_scope._tasks.discard(host_task)
 436  
 437          self._timeout()
 438          self._active = True
 439  
 440          # Start cancelling the host task if the scope was cancelled before entering
 441          if self._cancel_called:
 442              self._deliver_cancellation(self)
 443  
 444          return self
 445  
 446      def __exit__(
 447          self,
 448          exc_type: type[BaseException] | None,
 449          exc_val: BaseException | None,
 450          exc_tb: TracebackType | None,
 451      ) -> bool:
 452          del exc_tb
 453  
 454          if not self._active:
 455              raise RuntimeError("This cancel scope is not active")
 456          if current_task() is not self._host_task:
 457              raise RuntimeError(
 458                  "Attempted to exit cancel scope in a different task than it was "
 459                  "entered in"
 460              )
 461  
 462          assert self._host_task is not None
 463          host_task_state = _task_states.get(self._host_task)
 464          if host_task_state is None or host_task_state.cancel_scope is not self:
 465              raise RuntimeError(
 466                  "Attempted to exit a cancel scope that isn't the current tasks's "
 467                  "current cancel scope"
 468              )
 469  
 470          try:
 471              self._active = False
 472              if self._timeout_handle:
 473                  self._timeout_handle.cancel()
 474                  self._timeout_handle = None
 475  
 476              self._tasks.remove(self._host_task)
 477              if self._parent_scope is not None:
 478                  self._parent_scope._child_scopes.remove(self)
 479                  self._parent_scope._tasks.add(self._host_task)
 480  
 481              host_task_state.cancel_scope = self._parent_scope
 482  
 483              # Restart the cancellation effort in the closest visible, cancelled parent
 484              # scope if necessary
 485              self._restart_cancellation_in_parent()
 486  
 487              # We only swallow the exception iff it was an AnyIO CancelledError, either
 488              # directly as exc_val or inside an exception group and there are no cancelled
 489              # parent cancel scopes visible to us here
 490              if self._cancel_called and not self._parent_cancellation_is_visible_to_us:
 491                  # For each level-cancel() call made on the host task, call uncancel()
 492                  while self._pending_uncancellations:
 493                      self._host_task.uncancel()
 494                      self._pending_uncancellations -= 1
 495  
 496                  # Update cancelled_caught and check for exceptions we must not swallow
 497                  cannot_swallow_exc_val = False
 498                  if exc_val is not None:
 499                      for exc in iterate_exceptions(exc_val):
 500                          if isinstance(exc, CancelledError) and is_anyio_cancellation(
 501                              exc
 502                          ):
 503                              self._cancelled_caught = True
 504                          else:
 505                              cannot_swallow_exc_val = True
 506  
 507                  return self._cancelled_caught and not cannot_swallow_exc_val
 508              else:
 509                  if self._pending_uncancellations:
 510                      assert self._parent_scope is not None
 511                      assert self._parent_scope._pending_uncancellations is not None
 512                      self._parent_scope._pending_uncancellations += (
 513                          self._pending_uncancellations
 514                      )
 515                      self._pending_uncancellations = 0
 516  
 517                  return False
 518          finally:
 519              self._host_task = None
 520              del exc_val
 521  
 522      @property
 523      def _effectively_cancelled(self) -> bool:
 524          cancel_scope: CancelScope | None = self
 525          while cancel_scope is not None:
 526              if cancel_scope._cancel_called:
 527                  return True
 528  
 529              if cancel_scope.shield:
 530                  return False
 531  
 532              cancel_scope = cancel_scope._parent_scope
 533  
 534          return False
 535  
 536      @property
 537      def _parent_cancellation_is_visible_to_us(self) -> bool:
 538          return (
 539              self._parent_scope is not None
 540              and not self.shield
 541              and self._parent_scope._effectively_cancelled
 542          )
 543  
 544      def _timeout(self) -> None:
 545          if self._deadline != math.inf:
 546              loop = get_running_loop()
 547              if loop.time() >= self._deadline:
 548                  self.cancel()
 549              else:
 550                  self._timeout_handle = loop.call_at(self._deadline, self._timeout)
 551  
 552      def _deliver_cancellation(self, origin: CancelScope) -> bool:
 553          """
 554          Deliver cancellation to directly contained tasks and nested cancel scopes.
 555  
 556          Schedule another run at the end if we still have tasks eligible for
 557          cancellation.
 558  
 559          :param origin: the cancel scope that originated the cancellation
 560          :return: ``True`` if the delivery needs to be retried on the next cycle
 561  
 562          """
 563          should_retry = False
 564          current = current_task()
 565          for task in self._tasks:
 566              should_retry = True
 567              if task._must_cancel:  # type: ignore[attr-defined]
 568                  continue
 569  
 570              # The task is eligible for cancellation if it has started
 571              if task is not current and (task is self._host_task or _task_started(task)):
 572                  waiter = task._fut_waiter  # type: ignore[attr-defined]
 573                  if not isinstance(waiter, asyncio.Future) or not waiter.done():
 574                      task.cancel(f"Cancelled by cancel scope {id(origin):x}")
 575                      if (
 576                          task is origin._host_task
 577                          and origin._pending_uncancellations is not None
 578                      ):
 579                          origin._pending_uncancellations += 1
 580  
 581          # Deliver cancellation to child scopes that aren't shielded or running their own
 582          # cancellation callbacks
 583          for scope in self._child_scopes:
 584              if not scope._shield and not scope.cancel_called:
 585                  should_retry = scope._deliver_cancellation(origin) or should_retry
 586  
 587          # Schedule another callback if there are still tasks left
 588          if origin is self:
 589              if should_retry:
 590                  self._cancel_handle = get_running_loop().call_soon(
 591                      self._deliver_cancellation, origin
 592                  )
 593              else:
 594                  self._cancel_handle = None
 595  
 596          return should_retry
 597  
 598      def _restart_cancellation_in_parent(self) -> None:
 599          """
 600          Restart the cancellation effort in the closest directly cancelled parent scope.
 601  
 602          """
 603          scope = self._parent_scope
 604          while scope is not None:
 605              if scope._cancel_called:
 606                  if scope._cancel_handle is None:
 607                      scope._deliver_cancellation(scope)
 608  
 609                  break
 610  
 611              # No point in looking beyond any shielded scope
 612              if scope._shield:
 613                  break
 614  
 615              scope = scope._parent_scope
 616  
 617      def cancel(self) -> None:
 618          if not self._cancel_called:
 619              if self._timeout_handle:
 620                  self._timeout_handle.cancel()
 621                  self._timeout_handle = None
 622  
 623              self._cancel_called = True
 624              if self._host_task is not None:
 625                  self._deliver_cancellation(self)
 626  
 627      @property
 628      def deadline(self) -> float:
 629          return self._deadline
 630  
 631      @deadline.setter
 632      def deadline(self, value: float) -> None:
 633          self._deadline = float(value)
 634          if self._timeout_handle is not None:
 635              self._timeout_handle.cancel()
 636              self._timeout_handle = None
 637  
 638          if self._active and not self._cancel_called:
 639              self._timeout()
 640  
 641      @property
 642      def cancel_called(self) -> bool:
 643          return self._cancel_called
 644  
 645      @property
 646      def cancelled_caught(self) -> bool:
 647          return self._cancelled_caught
 648  
 649      @property
 650      def shield(self) -> bool:
 651          return self._shield
 652  
 653      @shield.setter
 654      def shield(self, value: bool) -> None:
 655          if self._shield != value:
 656              self._shield = value
 657              if not value:
 658                  self._restart_cancellation_in_parent()
 659  
 660  
 661  #
 662  # Task states
 663  #
 664  
 665  
 666  class TaskState:
 667      """
 668      Encapsulates auxiliary task information that cannot be added to the Task instance
 669      itself because there are no guarantees about its implementation.
 670      """
 671  
 672      __slots__ = "parent_id", "cancel_scope", "__weakref__"
 673  
 674      def __init__(self, parent_id: int | None, cancel_scope: CancelScope | None):
 675          self.parent_id = parent_id
 676          self.cancel_scope = cancel_scope
 677  
 678  
 679  _task_states: WeakKeyDictionary[asyncio.Task, TaskState] = WeakKeyDictionary()
 680  
 681  
 682  #
 683  # Task groups
 684  #
 685  
 686  
 687  class _AsyncioTaskStatus(abc.TaskStatus):
 688      def __init__(self, future: asyncio.Future, parent_id: int):
 689          self._future = future
 690          self._parent_id = parent_id
 691  
 692      def started(self, value: T_contra | None = None) -> None:
 693          try:
 694              self._future.set_result(value)
 695          except asyncio.InvalidStateError:
 696              if not self._future.cancelled():
 697                  raise RuntimeError(
 698                      "called 'started' twice on the same task status"
 699                  ) from None
 700  
 701          task = cast(asyncio.Task, current_task())
 702          _task_states[task].parent_id = self._parent_id
 703  
 704  
 705  if sys.version_info >= (3, 12):
 706      _eager_task_factory_code: CodeType | None = asyncio.eager_task_factory.__code__
 707  else:
 708      _eager_task_factory_code = None
 709  
 710  
 711  class TaskGroup(abc.TaskGroup):
 712      def __init__(self) -> None:
 713          self.cancel_scope: CancelScope = CancelScope()
 714          self._active = False
 715          self._exceptions: list[BaseException] = []
 716          self._tasks: set[asyncio.Task] = set()
 717          self._on_completed_fut: asyncio.Future[None] | None = None
 718  
 719      async def __aenter__(self) -> TaskGroup:
 720          self.cancel_scope.__enter__()
 721          self._active = True
 722          return self
 723  
 724      async def __aexit__(
 725          self,
 726          exc_type: type[BaseException] | None,
 727          exc_val: BaseException | None,
 728          exc_tb: TracebackType | None,
 729      ) -> bool | None:
 730          try:
 731              if exc_val is not None:
 732                  self.cancel_scope.cancel()
 733                  if not isinstance(exc_val, CancelledError):
 734                      self._exceptions.append(exc_val)
 735  
 736              loop = get_running_loop()
 737              try:
 738                  if self._tasks:
 739                      with CancelScope() as wait_scope:
 740                          while self._tasks:
 741                              self._on_completed_fut = loop.create_future()
 742  
 743                              try:
 744                                  await self._on_completed_fut
 745                              except CancelledError as exc:
 746                                  # Shield the scope against further cancellation attempts,
 747                                  # as they're not productive (#695)
 748                                  wait_scope.shield = True
 749                                  self.cancel_scope.cancel()
 750  
 751                                  # Set exc_val from the cancellation exception if it was
 752                                  # previously unset. However, we should not replace a native
 753                                  # cancellation exception with one raise by a cancel scope.
 754                                  if exc_val is None or (
 755                                      isinstance(exc_val, CancelledError)
 756                                      and not is_anyio_cancellation(exc)
 757                                  ):
 758                                      exc_val = exc
 759  
 760                              self._on_completed_fut = None
 761                  else:
 762                      # If there are no child tasks to wait on, run at least one checkpoint
 763                      # anyway
 764                      await AsyncIOBackend.cancel_shielded_checkpoint()
 765  
 766                  self._active = False
 767                  if self._exceptions:
 768                      # The exception that got us here should already have been
 769                      # added to self._exceptions so it's ok to break exception
 770                      # chaining and avoid adding a "During handling of above..."
 771                      # for each nesting level.
 772                      raise BaseExceptionGroup(
 773                          "unhandled errors in a TaskGroup", self._exceptions
 774                      ) from None
 775                  elif exc_val:
 776                      raise exc_val
 777              except BaseException as exc:
 778                  if self.cancel_scope.__exit__(type(exc), exc, exc.__traceback__):
 779                      return True
 780  
 781                  raise
 782  
 783              return self.cancel_scope.__exit__(exc_type, exc_val, exc_tb)
 784          finally:
 785              del exc_val, exc_tb, self._exceptions
 786  
 787      def _spawn(
 788          self,
 789          func: Callable[[Unpack[PosArgsT]], Awaitable[Any]],
 790          args: tuple[Unpack[PosArgsT]],
 791          name: object,
 792          task_status_future: asyncio.Future | None = None,
 793      ) -> asyncio.Task:
 794          def task_done(_task: asyncio.Task) -> None:
 795              task_state = _task_states[_task]
 796              assert task_state.cancel_scope is not None
 797              assert _task in task_state.cancel_scope._tasks
 798              task_state.cancel_scope._tasks.remove(_task)
 799              self._tasks.remove(task)
 800              del _task_states[_task]
 801  
 802              if self._on_completed_fut is not None and not self._tasks:
 803                  try:
 804                      self._on_completed_fut.set_result(None)
 805                  except asyncio.InvalidStateError:
 806                      pass
 807  
 808              try:
 809                  exc = _task.exception()
 810              except CancelledError as e:
 811                  while isinstance(e.__context__, CancelledError):
 812                      e = e.__context__
 813  
 814                  exc = e
 815  
 816              if exc is not None:
 817                  # The future can only be in the cancelled state if the host task was
 818                  # cancelled, so return immediately instead of adding one more
 819                  # CancelledError to the exceptions list
 820                  if task_status_future is not None and task_status_future.cancelled():
 821                      return
 822  
 823                  if task_status_future is None or task_status_future.done():
 824                      if not isinstance(exc, CancelledError):
 825                          self._exceptions.append(exc)
 826  
 827                      if not self.cancel_scope._effectively_cancelled:
 828                          self.cancel_scope.cancel()
 829                  else:
 830                      task_status_future.set_exception(exc)
 831              elif task_status_future is not None and not task_status_future.done():
 832                  task_status_future.set_exception(
 833                      RuntimeError("Child exited without calling task_status.started()")
 834                  )
 835  
 836          if not self._active:
 837              raise RuntimeError(
 838                  "This task group is not active; no new tasks can be started."
 839              )
 840  
 841          kwargs = {}
 842          if task_status_future:
 843              parent_id = id(current_task())
 844              kwargs["task_status"] = _AsyncioTaskStatus(
 845                  task_status_future, id(self.cancel_scope._host_task)
 846              )
 847          else:
 848              parent_id = id(self.cancel_scope._host_task)
 849  
 850          coro = func(*args, **kwargs)
 851          if not iscoroutine(coro):
 852              prefix = f"{func.__module__}." if hasattr(func, "__module__") else ""
 853              raise TypeError(
 854                  f"Expected {prefix}{func.__qualname__}() to return a coroutine, but "
 855                  f"the return value ({coro!r}) is not a coroutine object"
 856              )
 857  
 858          name = get_callable_name(func) if name is None else str(name)
 859          loop = asyncio.get_running_loop()
 860          if (
 861              (factory := loop.get_task_factory())
 862              and getattr(factory, "__code__", None) is _eager_task_factory_code
 863              and (closure := getattr(factory, "__closure__", None))
 864          ):
 865              custom_task_constructor = closure[0].cell_contents
 866              task = custom_task_constructor(coro, loop=loop, name=name)
 867          else:
 868              task = create_task(coro, name=name)
 869  
 870          # Make the spawned task inherit the task group's cancel scope
 871          _task_states[task] = TaskState(
 872              parent_id=parent_id, cancel_scope=self.cancel_scope
 873          )
 874          self.cancel_scope._tasks.add(task)
 875          self._tasks.add(task)
 876          task.add_done_callback(task_done)
 877          return task
 878  
 879      def start_soon(
 880          self,
 881          func: Callable[[Unpack[PosArgsT]], Awaitable[Any]],
 882          *args: Unpack[PosArgsT],
 883          name: object = None,
 884      ) -> None:
 885          self._spawn(func, args, name)
 886  
 887      async def start(
 888          self, func: Callable[..., Awaitable[Any]], *args: object, name: object = None
 889      ) -> Any:
 890          future: asyncio.Future = asyncio.Future()
 891          task = self._spawn(func, args, name, future)
 892  
 893          # If the task raises an exception after sending a start value without a switch
 894          # point between, the task group is cancelled and this method never proceeds to
 895          # process the completed future. That's why we have to have a shielded cancel
 896          # scope here.
 897          try:
 898              return await future
 899          except CancelledError:
 900              # Cancel the task and wait for it to exit before returning
 901              task.cancel()
 902              with CancelScope(shield=True), suppress(CancelledError):
 903                  await task
 904  
 905              raise
 906  
 907  
 908  #
 909  # Threads
 910  #
 911  
 912  _Retval_Queue_Type = tuple[Optional[T_Retval], Optional[BaseException]]
 913  
 914  
 915  class WorkerThread(Thread):
 916      MAX_IDLE_TIME = 10  # seconds
 917  
 918      def __init__(
 919          self,
 920          root_task: asyncio.Task,
 921          workers: set[WorkerThread],
 922          idle_workers: deque[WorkerThread],
 923      ):
 924          super().__init__(name="AnyIO worker thread")
 925          self.root_task = root_task
 926          self.workers = workers
 927          self.idle_workers = idle_workers
 928          self.loop = root_task._loop
 929          self.queue: Queue[
 930              tuple[Context, Callable, tuple, asyncio.Future, CancelScope] | None
 931          ] = Queue(2)
 932          self.idle_since = AsyncIOBackend.current_time()
 933          self.stopping = False
 934  
 935      def _report_result(
 936          self, future: asyncio.Future, result: Any, exc: BaseException | None
 937      ) -> None:
 938          self.idle_since = AsyncIOBackend.current_time()
 939          if not self.stopping:
 940              self.idle_workers.append(self)
 941  
 942          if not future.cancelled():
 943              if exc is not None:
 944                  if isinstance(exc, StopIteration):
 945                      new_exc = RuntimeError("coroutine raised StopIteration")
 946                      new_exc.__cause__ = exc
 947                      exc = new_exc
 948  
 949                  future.set_exception(exc)
 950              else:
 951                  future.set_result(result)
 952  
 953      def run(self) -> None:
 954          with claim_worker_thread(AsyncIOBackend, self.loop):
 955              while True:
 956                  item = self.queue.get()
 957                  if item is None:
 958                      # Shutdown command received
 959                      return
 960  
 961                  context, func, args, future, cancel_scope = item
 962                  if not future.cancelled():
 963                      result = None
 964                      exception: BaseException | None = None
 965                      threadlocals.current_cancel_scope = cancel_scope
 966                      try:
 967                          result = context.run(func, *args)
 968                      except BaseException as exc:
 969                          exception = exc
 970                      finally:
 971                          del threadlocals.current_cancel_scope
 972  
 973                      if not self.loop.is_closed():
 974                          self.loop.call_soon_threadsafe(
 975                              self._report_result, future, result, exception
 976                          )
 977  
 978                      del result, exception
 979  
 980                  self.queue.task_done()
 981                  del item, context, func, args, future, cancel_scope
 982  
 983      def stop(self, f: asyncio.Task | None = None) -> None:
 984          self.stopping = True
 985          self.queue.put_nowait(None)
 986          self.workers.discard(self)
 987          try:
 988              self.idle_workers.remove(self)
 989          except ValueError:
 990              pass
 991  
 992  
 993  _threadpool_idle_workers: RunVar[deque[WorkerThread]] = RunVar(
 994      "_threadpool_idle_workers"
 995  )
 996  _threadpool_workers: RunVar[set[WorkerThread]] = RunVar("_threadpool_workers")
 997  
 998  
 999  class BlockingPortal(abc.BlockingPortal):
1000      def __new__(cls) -> BlockingPortal:
1001          return object.__new__(cls)
1002  
1003      def __init__(self) -> None:
1004          super().__init__()
1005          self._loop = get_running_loop()
1006  
1007      def _spawn_task_from_thread(
1008          self,
1009          func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
1010          args: tuple[Unpack[PosArgsT]],
1011          kwargs: dict[str, Any],
1012          name: object,
1013          future: Future[T_Retval],
1014      ) -> None:
1015          AsyncIOBackend.run_sync_from_thread(
1016              partial(self._task_group.start_soon, name=name),
1017              (self._call_func, func, args, kwargs, future),
1018              self._loop,
1019          )
1020  
1021  
1022  #
1023  # Subprocesses
1024  #
1025  
1026  
1027  @dataclass(eq=False)
1028  class StreamReaderWrapper(abc.ByteReceiveStream):
1029      _stream: asyncio.StreamReader
1030  
1031      async def receive(self, max_bytes: int = 65536) -> bytes:
1032          data = await self._stream.read(max_bytes)
1033          if data:
1034              return data
1035          else:
1036              raise EndOfStream
1037  
1038      async def aclose(self) -> None:
1039          self._stream.set_exception(ClosedResourceError())
1040          await AsyncIOBackend.checkpoint()
1041  
1042  
1043  @dataclass(eq=False)
1044  class StreamWriterWrapper(abc.ByteSendStream):
1045      _stream: asyncio.StreamWriter
1046  
1047      async def send(self, item: bytes) -> None:
1048          self._stream.write(item)
1049          await self._stream.drain()
1050  
1051      async def aclose(self) -> None:
1052          self._stream.close()
1053          await AsyncIOBackend.checkpoint()
1054  
1055  
1056  @dataclass(eq=False)
1057  class Process(abc.Process):
1058      _process: asyncio.subprocess.Process
1059      _stdin: StreamWriterWrapper | None
1060      _stdout: StreamReaderWrapper | None
1061      _stderr: StreamReaderWrapper | None
1062  
1063      async def aclose(self) -> None:
1064          with CancelScope(shield=True) as scope:
1065              if self._stdin:
1066                  await self._stdin.aclose()
1067              if self._stdout:
1068                  await self._stdout.aclose()
1069              if self._stderr:
1070                  await self._stderr.aclose()
1071  
1072              scope.shield = False
1073              try:
1074                  await self.wait()
1075              except BaseException:
1076                  scope.shield = True
1077                  self.kill()
1078                  await self.wait()
1079                  raise
1080  
1081      async def wait(self) -> int:
1082          return await self._process.wait()
1083  
1084      def terminate(self) -> None:
1085          self._process.terminate()
1086  
1087      def kill(self) -> None:
1088          self._process.kill()
1089  
1090      def send_signal(self, signal: int) -> None:
1091          self._process.send_signal(signal)
1092  
1093      @property
1094      def pid(self) -> int:
1095          return self._process.pid
1096  
1097      @property
1098      def returncode(self) -> int | None:
1099          return self._process.returncode
1100  
1101      @property
1102      def stdin(self) -> abc.ByteSendStream | None:
1103          return self._stdin
1104  
1105      @property
1106      def stdout(self) -> abc.ByteReceiveStream | None:
1107          return self._stdout
1108  
1109      @property
1110      def stderr(self) -> abc.ByteReceiveStream | None:
1111          return self._stderr
1112  
1113  
1114  def _forcibly_shutdown_process_pool_on_exit(
1115      workers: set[Process], _task: object
1116  ) -> None:
1117      """
1118      Forcibly shuts down worker processes belonging to this event loop."""
1119      child_watcher: asyncio.AbstractChildWatcher | None = None
1120      if sys.version_info < (3, 12):
1121          try:
1122              child_watcher = asyncio.get_event_loop_policy().get_child_watcher()
1123          except NotImplementedError:
1124              pass
1125  
1126      # Close as much as possible (w/o async/await) to avoid warnings
1127      for process in workers:
1128          if process.returncode is None:
1129              continue
1130  
1131          process._stdin._stream._transport.close()  # type: ignore[union-attr]
1132          process._stdout._stream._transport.close()  # type: ignore[union-attr]
1133          process._stderr._stream._transport.close()  # type: ignore[union-attr]
1134          process.kill()
1135          if child_watcher:
1136              child_watcher.remove_child_handler(process.pid)
1137  
1138  
1139  async def _shutdown_process_pool_on_exit(workers: set[abc.Process]) -> None:
1140      """
1141      Shuts down worker processes belonging to this event loop.
1142  
1143      NOTE: this only works when the event loop was started using asyncio.run() or
1144      anyio.run().
1145  
1146      """
1147      process: abc.Process
1148      try:
1149          await sleep(math.inf)
1150      except asyncio.CancelledError:
1151          for process in workers:
1152              if process.returncode is None:
1153                  process.kill()
1154  
1155          for process in workers:
1156              await process.aclose()
1157  
1158  
1159  #
1160  # Sockets and networking
1161  #
1162  
1163  
1164  class StreamProtocol(asyncio.Protocol):
1165      read_queue: deque[bytes]
1166      read_event: asyncio.Event
1167      write_event: asyncio.Event
1168      exception: Exception | None = None
1169      is_at_eof: bool = False
1170  
1171      def connection_made(self, transport: asyncio.BaseTransport) -> None:
1172          self.read_queue = deque()
1173          self.read_event = asyncio.Event()
1174          self.write_event = asyncio.Event()
1175          self.write_event.set()
1176          cast(asyncio.Transport, transport).set_write_buffer_limits(0)
1177  
1178      def connection_lost(self, exc: Exception | None) -> None:
1179          if exc:
1180              self.exception = BrokenResourceError()
1181              self.exception.__cause__ = exc
1182  
1183          self.read_event.set()
1184          self.write_event.set()
1185  
1186      def data_received(self, data: bytes) -> None:
1187          # ProactorEventloop sometimes sends bytearray instead of bytes
1188          self.read_queue.append(bytes(data))
1189          self.read_event.set()
1190  
1191      def eof_received(self) -> bool | None:
1192          self.is_at_eof = True
1193          self.read_event.set()
1194          return True
1195  
1196      def pause_writing(self) -> None:
1197          self.write_event = asyncio.Event()
1198  
1199      def resume_writing(self) -> None:
1200          self.write_event.set()
1201  
1202  
1203  class DatagramProtocol(asyncio.DatagramProtocol):
1204      read_queue: deque[tuple[bytes, IPSockAddrType]]
1205      read_event: asyncio.Event
1206      write_event: asyncio.Event
1207      exception: Exception | None = None
1208  
1209      def connection_made(self, transport: asyncio.BaseTransport) -> None:
1210          self.read_queue = deque(maxlen=100)  # arbitrary value
1211          self.read_event = asyncio.Event()
1212          self.write_event = asyncio.Event()
1213          self.write_event.set()
1214  
1215      def connection_lost(self, exc: Exception | None) -> None:
1216          self.read_event.set()
1217          self.write_event.set()
1218  
1219      def datagram_received(self, data: bytes, addr: IPSockAddrType) -> None:
1220          addr = convert_ipv6_sockaddr(addr)
1221          self.read_queue.append((data, addr))
1222          self.read_event.set()
1223  
1224      def error_received(self, exc: Exception) -> None:
1225          self.exception = exc
1226  
1227      def pause_writing(self) -> None:
1228          self.write_event.clear()
1229  
1230      def resume_writing(self) -> None:
1231          self.write_event.set()
1232  
1233  
1234  class SocketStream(abc.SocketStream):
1235      def __init__(self, transport: asyncio.Transport, protocol: StreamProtocol):
1236          self._transport = transport
1237          self._protocol = protocol
1238          self._receive_guard = ResourceGuard("reading from")
1239          self._send_guard = ResourceGuard("writing to")
1240          self._closed = False
1241  
1242      @property
1243      def _raw_socket(self) -> socket.socket:
1244          return self._transport.get_extra_info("socket")
1245  
1246      async def receive(self, max_bytes: int = 65536) -> bytes:
1247          with self._receive_guard:
1248              if (
1249                  not self._protocol.read_event.is_set()
1250                  and not self._transport.is_closing()
1251                  and not self._protocol.is_at_eof
1252              ):
1253                  self._transport.resume_reading()
1254                  await self._protocol.read_event.wait()
1255                  self._transport.pause_reading()
1256              else:
1257                  await AsyncIOBackend.checkpoint()
1258  
1259              try:
1260                  chunk = self._protocol.read_queue.popleft()
1261              except IndexError:
1262                  if self._closed:
1263                      raise ClosedResourceError from None
1264                  elif self._protocol.exception:
1265                      raise self._protocol.exception from None
1266                  else:
1267                      raise EndOfStream from None
1268  
1269              if len(chunk) > max_bytes:
1270                  # Split the oversized chunk
1271                  chunk, leftover = chunk[:max_bytes], chunk[max_bytes:]
1272                  self._protocol.read_queue.appendleft(leftover)
1273  
1274              # If the read queue is empty, clear the flag so that the next call will
1275              # block until data is available
1276              if not self._protocol.read_queue:
1277                  self._protocol.read_event.clear()
1278  
1279          return chunk
1280  
1281      async def send(self, item: bytes) -> None:
1282          with self._send_guard:
1283              await AsyncIOBackend.checkpoint()
1284  
1285              if self._closed:
1286                  raise ClosedResourceError
1287              elif self._protocol.exception is not None:
1288                  raise self._protocol.exception
1289  
1290              try:
1291                  self._transport.write(item)
1292              except RuntimeError as exc:
1293                  if self._transport.is_closing():
1294                      raise BrokenResourceError from exc
1295                  else:
1296                      raise
1297  
1298              await self._protocol.write_event.wait()
1299  
1300      async def send_eof(self) -> None:
1301          try:
1302              self._transport.write_eof()
1303          except OSError:
1304              pass
1305  
1306      async def aclose(self) -> None:
1307          if not self._transport.is_closing():
1308              self._closed = True
1309              try:
1310                  self._transport.write_eof()
1311              except OSError:
1312                  pass
1313  
1314              self._transport.close()
1315              await sleep(0)
1316              self._transport.abort()
1317  
1318  
1319  class _RawSocketMixin:
1320      _receive_future: asyncio.Future | None = None
1321      _send_future: asyncio.Future | None = None
1322      _closing = False
1323  
1324      def __init__(self, raw_socket: socket.socket):
1325          self.__raw_socket = raw_socket
1326          self._receive_guard = ResourceGuard("reading from")
1327          self._send_guard = ResourceGuard("writing to")
1328  
1329      @property
1330      def _raw_socket(self) -> socket.socket:
1331          return self.__raw_socket
1332  
1333      def _wait_until_readable(self, loop: asyncio.AbstractEventLoop) -> asyncio.Future:
1334          def callback(f: object) -> None:
1335              del self._receive_future
1336              loop.remove_reader(self.__raw_socket)
1337  
1338          f = self._receive_future = asyncio.Future()
1339          loop.add_reader(self.__raw_socket, f.set_result, None)
1340          f.add_done_callback(callback)
1341          return f
1342  
1343      def _wait_until_writable(self, loop: asyncio.AbstractEventLoop) -> asyncio.Future:
1344          def callback(f: object) -> None:
1345              del self._send_future
1346              loop.remove_writer(self.__raw_socket)
1347  
1348          f = self._send_future = asyncio.Future()
1349          loop.add_writer(self.__raw_socket, f.set_result, None)
1350          f.add_done_callback(callback)
1351          return f
1352  
1353      async def aclose(self) -> None:
1354          if not self._closing:
1355              self._closing = True
1356              if self.__raw_socket.fileno() != -1:
1357                  self.__raw_socket.close()
1358  
1359              if self._receive_future:
1360                  self._receive_future.set_result(None)
1361              if self._send_future:
1362                  self._send_future.set_result(None)
1363  
1364  
1365  class UNIXSocketStream(_RawSocketMixin, abc.UNIXSocketStream):
1366      async def send_eof(self) -> None:
1367          with self._send_guard:
1368              self._raw_socket.shutdown(socket.SHUT_WR)
1369  
1370      async def receive(self, max_bytes: int = 65536) -> bytes:
1371          loop = get_running_loop()
1372          await AsyncIOBackend.checkpoint()
1373          with self._receive_guard:
1374              while True:
1375                  try:
1376                      data = self._raw_socket.recv(max_bytes)
1377                  except BlockingIOError:
1378                      await self._wait_until_readable(loop)
1379                  except OSError as exc:
1380                      if self._closing:
1381                          raise ClosedResourceError from None
1382                      else:
1383                          raise BrokenResourceError from exc
1384                  else:
1385                      if not data:
1386                          raise EndOfStream
1387  
1388                      return data
1389  
1390      async def send(self, item: bytes) -> None:
1391          loop = get_running_loop()
1392          await AsyncIOBackend.checkpoint()
1393          with self._send_guard:
1394              view = memoryview(item)
1395              while view:
1396                  try:
1397                      bytes_sent = self._raw_socket.send(view)
1398                  except BlockingIOError:
1399                      await self._wait_until_writable(loop)
1400                  except OSError as exc:
1401                      if self._closing:
1402                          raise ClosedResourceError from None
1403                      else:
1404                          raise BrokenResourceError from exc
1405                  else:
1406                      view = view[bytes_sent:]
1407  
1408      async def receive_fds(self, msglen: int, maxfds: int) -> tuple[bytes, list[int]]:
1409          if not isinstance(msglen, int) or msglen < 0:
1410              raise ValueError("msglen must be a non-negative integer")
1411          if not isinstance(maxfds, int) or maxfds < 1:
1412              raise ValueError("maxfds must be a positive integer")
1413  
1414          loop = get_running_loop()
1415          fds = array.array("i")
1416          await AsyncIOBackend.checkpoint()
1417          with self._receive_guard:
1418              while True:
1419                  try:
1420                      message, ancdata, flags, addr = self._raw_socket.recvmsg(
1421                          msglen, socket.CMSG_LEN(maxfds * fds.itemsize)
1422                      )
1423                  except BlockingIOError:
1424                      await self._wait_until_readable(loop)
1425                  except OSError as exc:
1426                      if self._closing:
1427                          raise ClosedResourceError from None
1428                      else:
1429                          raise BrokenResourceError from exc
1430                  else:
1431                      if not message and not ancdata:
1432                          raise EndOfStream
1433  
1434                      break
1435  
1436          for cmsg_level, cmsg_type, cmsg_data in ancdata:
1437              if cmsg_level != socket.SOL_SOCKET or cmsg_type != socket.SCM_RIGHTS:
1438                  raise RuntimeError(
1439                      f"Received unexpected ancillary data; message = {message!r}, "
1440                      f"cmsg_level = {cmsg_level}, cmsg_type = {cmsg_type}"
1441                  )
1442  
1443              fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
1444  
1445          return message, list(fds)
1446  
1447      async def send_fds(self, message: bytes, fds: Collection[int | IOBase]) -> None:
1448          if not message:
1449              raise ValueError("message must not be empty")
1450          if not fds:
1451              raise ValueError("fds must not be empty")
1452  
1453          loop = get_running_loop()
1454          filenos: list[int] = []
1455          for fd in fds:
1456              if isinstance(fd, int):
1457                  filenos.append(fd)
1458              elif isinstance(fd, IOBase):
1459                  filenos.append(fd.fileno())
1460  
1461          fdarray = array.array("i", filenos)
1462          await AsyncIOBackend.checkpoint()
1463          with self._send_guard:
1464              while True:
1465                  try:
1466                      # The ignore can be removed after mypy picks up
1467                      # https://github.com/python/typeshed/pull/5545
1468                      self._raw_socket.sendmsg(
1469                          [message], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fdarray)]
1470                      )
1471                      break
1472                  except BlockingIOError:
1473                      await self._wait_until_writable(loop)
1474                  except OSError as exc:
1475                      if self._closing:
1476                          raise ClosedResourceError from None
1477                      else:
1478                          raise BrokenResourceError from exc
1479  
1480  
1481  class TCPSocketListener(abc.SocketListener):
1482      _accept_scope: CancelScope | None = None
1483      _closed = False
1484  
1485      def __init__(self, raw_socket: socket.socket):
1486          self.__raw_socket = raw_socket
1487          self._loop = cast(asyncio.BaseEventLoop, get_running_loop())
1488          self._accept_guard = ResourceGuard("accepting connections from")
1489  
1490      @property
1491      def _raw_socket(self) -> socket.socket:
1492          return self.__raw_socket
1493  
1494      async def accept(self) -> abc.SocketStream:
1495          if self._closed:
1496              raise ClosedResourceError
1497  
1498          with self._accept_guard:
1499              await AsyncIOBackend.checkpoint()
1500              with CancelScope() as self._accept_scope:
1501                  try:
1502                      client_sock, _addr = await self._loop.sock_accept(self._raw_socket)
1503                  except asyncio.CancelledError:
1504                      # Workaround for https://bugs.python.org/issue41317
1505                      try:
1506                          self._loop.remove_reader(self._raw_socket)
1507                      except (ValueError, NotImplementedError):
1508                          pass
1509  
1510                      if self._closed:
1511                          raise ClosedResourceError from None
1512  
1513                      raise
1514                  finally:
1515                      self._accept_scope = None
1516  
1517          client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
1518          transport, protocol = await self._loop.connect_accepted_socket(
1519              StreamProtocol, client_sock
1520          )
1521          return SocketStream(transport, protocol)
1522  
1523      async def aclose(self) -> None:
1524          if self._closed:
1525              return
1526  
1527          self._closed = True
1528          if self._accept_scope:
1529              # Workaround for https://bugs.python.org/issue41317
1530              try:
1531                  self._loop.remove_reader(self._raw_socket)
1532              except (ValueError, NotImplementedError):
1533                  pass
1534  
1535              self._accept_scope.cancel()
1536              await sleep(0)
1537  
1538          self._raw_socket.close()
1539  
1540  
1541  class UNIXSocketListener(abc.SocketListener):
1542      def __init__(self, raw_socket: socket.socket):
1543          self.__raw_socket = raw_socket
1544          self._loop = get_running_loop()
1545          self._accept_guard = ResourceGuard("accepting connections from")
1546          self._closed = False
1547  
1548      async def accept(self) -> abc.SocketStream:
1549          await AsyncIOBackend.checkpoint()
1550          with self._accept_guard:
1551              while True:
1552                  try:
1553                      client_sock, _ = self.__raw_socket.accept()
1554                      client_sock.setblocking(False)
1555                      return UNIXSocketStream(client_sock)
1556                  except BlockingIOError:
1557                      f: asyncio.Future = asyncio.Future()
1558                      self._loop.add_reader(self.__raw_socket, f.set_result, None)
1559                      f.add_done_callback(
1560                          lambda _: self._loop.remove_reader(self.__raw_socket)
1561                      )
1562                      await f
1563                  except OSError as exc:
1564                      if self._closed:
1565                          raise ClosedResourceError from None
1566                      else:
1567                          raise BrokenResourceError from exc
1568  
1569      async def aclose(self) -> None:
1570          self._closed = True
1571          self.__raw_socket.close()
1572  
1573      @property
1574      def _raw_socket(self) -> socket.socket:
1575          return self.__raw_socket
1576  
1577  
1578  class UDPSocket(abc.UDPSocket):
1579      def __init__(
1580          self, transport: asyncio.DatagramTransport, protocol: DatagramProtocol
1581      ):
1582          self._transport = transport
1583          self._protocol = protocol
1584          self._receive_guard = ResourceGuard("reading from")
1585          self._send_guard = ResourceGuard("writing to")
1586          self._closed = False
1587  
1588      @property
1589      def _raw_socket(self) -> socket.socket:
1590          return self._transport.get_extra_info("socket")
1591  
1592      async def aclose(self) -> None:
1593          if not self._transport.is_closing():
1594              self._closed = True
1595              self._transport.close()
1596  
1597      async def receive(self) -> tuple[bytes, IPSockAddrType]:
1598          with self._receive_guard:
1599              await AsyncIOBackend.checkpoint()
1600  
1601              # If the buffer is empty, ask for more data
1602              if not self._protocol.read_queue and not self._transport.is_closing():
1603                  self._protocol.read_event.clear()
1604                  await self._protocol.read_event.wait()
1605  
1606              try:
1607                  return self._protocol.read_queue.popleft()
1608              except IndexError:
1609                  if self._closed:
1610                      raise ClosedResourceError from None
1611                  else:
1612                      raise BrokenResourceError from None
1613  
1614      async def send(self, item: UDPPacketType) -> None:
1615          with self._send_guard:
1616              await AsyncIOBackend.checkpoint()
1617              await self._protocol.write_event.wait()
1618              if self._closed:
1619                  raise ClosedResourceError
1620              elif self._transport.is_closing():
1621                  raise BrokenResourceError
1622              else:
1623                  self._transport.sendto(*item)
1624  
1625  
1626  class ConnectedUDPSocket(abc.ConnectedUDPSocket):
1627      def __init__(
1628          self, transport: asyncio.DatagramTransport, protocol: DatagramProtocol
1629      ):
1630          self._transport = transport
1631          self._protocol = protocol
1632          self._receive_guard = ResourceGuard("reading from")
1633          self._send_guard = ResourceGuard("writing to")
1634          self._closed = False
1635  
1636      @property
1637      def _raw_socket(self) -> socket.socket:
1638          return self._transport.get_extra_info("socket")
1639  
1640      async def aclose(self) -> None:
1641          if not self._transport.is_closing():
1642              self._closed = True
1643              self._transport.close()
1644  
1645      async def receive(self) -> bytes:
1646          with self._receive_guard:
1647              await AsyncIOBackend.checkpoint()
1648  
1649              # If the buffer is empty, ask for more data
1650              if not self._protocol.read_queue and not self._transport.is_closing():
1651                  self._protocol.read_event.clear()
1652                  await self._protocol.read_event.wait()
1653  
1654              try:
1655                  packet = self._protocol.read_queue.popleft()
1656              except IndexError:
1657                  if self._closed:
1658                      raise ClosedResourceError from None
1659                  else:
1660                      raise BrokenResourceError from None
1661  
1662              return packet[0]
1663  
1664      async def send(self, item: bytes) -> None:
1665          with self._send_guard:
1666              await AsyncIOBackend.checkpoint()
1667              await self._protocol.write_event.wait()
1668              if self._closed:
1669                  raise ClosedResourceError
1670              elif self._transport.is_closing():
1671                  raise BrokenResourceError
1672              else:
1673                  self._transport.sendto(item)
1674  
1675  
1676  class UNIXDatagramSocket(_RawSocketMixin, abc.UNIXDatagramSocket):
1677      async def receive(self) -> UNIXDatagramPacketType:
1678          loop = get_running_loop()
1679          await AsyncIOBackend.checkpoint()
1680          with self._receive_guard:
1681              while True:
1682                  try:
1683                      data = self._raw_socket.recvfrom(65536)
1684                  except BlockingIOError:
1685                      await self._wait_until_readable(loop)
1686                  except OSError as exc:
1687                      if self._closing:
1688                          raise ClosedResourceError from None
1689                      else:
1690                          raise BrokenResourceError from exc
1691                  else:
1692                      return data
1693  
1694      async def send(self, item: UNIXDatagramPacketType) -> None:
1695          loop = get_running_loop()
1696          await AsyncIOBackend.checkpoint()
1697          with self._send_guard:
1698              while True:
1699                  try:
1700                      self._raw_socket.sendto(*item)
1701                  except BlockingIOError:
1702                      await self._wait_until_writable(loop)
1703                  except OSError as exc:
1704                      if self._closing:
1705                          raise ClosedResourceError from None
1706                      else:
1707                          raise BrokenResourceError from exc
1708                  else:
1709                      return
1710  
1711  
1712  class ConnectedUNIXDatagramSocket(_RawSocketMixin, abc.ConnectedUNIXDatagramSocket):
1713      async def receive(self) -> bytes:
1714          loop = get_running_loop()
1715          await AsyncIOBackend.checkpoint()
1716          with self._receive_guard:
1717              while True:
1718                  try:
1719                      data = self._raw_socket.recv(65536)
1720                  except BlockingIOError:
1721                      await self._wait_until_readable(loop)
1722                  except OSError as exc:
1723                      if self._closing:
1724                          raise ClosedResourceError from None
1725                      else:
1726                          raise BrokenResourceError from exc
1727                  else:
1728                      return data
1729  
1730      async def send(self, item: bytes) -> None:
1731          loop = get_running_loop()
1732          await AsyncIOBackend.checkpoint()
1733          with self._send_guard:
1734              while True:
1735                  try:
1736                      self._raw_socket.send(item)
1737                  except BlockingIOError:
1738                      await self._wait_until_writable(loop)
1739                  except OSError as exc:
1740                      if self._closing:
1741                          raise ClosedResourceError from None
1742                      else:
1743                          raise BrokenResourceError from exc
1744                  else:
1745                      return
1746  
1747  
1748  _read_events: RunVar[dict[int, asyncio.Event]] = RunVar("read_events")
1749  _write_events: RunVar[dict[int, asyncio.Event]] = RunVar("write_events")
1750  
1751  
1752  #
1753  # Synchronization
1754  #
1755  
1756  
1757  class Event(BaseEvent):
1758      def __new__(cls) -> Event:
1759          return object.__new__(cls)
1760  
1761      def __init__(self) -> None:
1762          self._event = asyncio.Event()
1763  
1764      def set(self) -> None:
1765          self._event.set()
1766  
1767      def is_set(self) -> bool:
1768          return self._event.is_set()
1769  
1770      async def wait(self) -> None:
1771          if self.is_set():
1772              await AsyncIOBackend.checkpoint()
1773          else:
1774              await self._event.wait()
1775  
1776      def statistics(self) -> EventStatistics:
1777          return EventStatistics(len(self._event._waiters))
1778  
1779  
1780  class Lock(BaseLock):
1781      def __new__(cls, *, fast_acquire: bool = False) -> Lock:
1782          return object.__new__(cls)
1783  
1784      def __init__(self, *, fast_acquire: bool = False) -> None:
1785          self._fast_acquire = fast_acquire
1786          self._owner_task: asyncio.Task | None = None
1787          self._waiters: deque[tuple[asyncio.Task, asyncio.Future]] = deque()
1788  
1789      async def acquire(self) -> None:
1790          task = cast(asyncio.Task, current_task())
1791          if self._owner_task is None and not self._waiters:
1792              await AsyncIOBackend.checkpoint_if_cancelled()
1793              self._owner_task = task
1794  
1795              # Unless on the "fast path", yield control of the event loop so that other
1796              # tasks can run too
1797              if not self._fast_acquire:
1798                  try:
1799                      await AsyncIOBackend.cancel_shielded_checkpoint()
1800                  except CancelledError:
1801                      self.release()
1802                      raise
1803  
1804              return
1805  
1806          if self._owner_task == task:
1807              raise RuntimeError("Attempted to acquire an already held Lock")
1808  
1809          fut: asyncio.Future[None] = asyncio.Future()
1810          item = task, fut
1811          self._waiters.append(item)
1812          try:
1813              await fut
1814          except CancelledError:
1815              self._waiters.remove(item)
1816              if self._owner_task is task:
1817                  self.release()
1818  
1819              raise
1820  
1821          self._waiters.remove(item)
1822  
1823      def acquire_nowait(self) -> None:
1824          task = cast(asyncio.Task, current_task())
1825          if self._owner_task is None and not self._waiters:
1826              self._owner_task = task
1827              return
1828  
1829          if self._owner_task is task:
1830              raise RuntimeError("Attempted to acquire an already held Lock")
1831  
1832          raise WouldBlock
1833  
1834      def locked(self) -> bool:
1835          return self._owner_task is not None
1836  
1837      def release(self) -> None:
1838          if self._owner_task != current_task():
1839              raise RuntimeError("The current task is not holding this lock")
1840  
1841          for task, fut in self._waiters:
1842              if not fut.cancelled():
1843                  self._owner_task = task
1844                  fut.set_result(None)
1845                  return
1846  
1847          self._owner_task = None
1848  
1849      def statistics(self) -> LockStatistics:
1850          task_info = AsyncIOTaskInfo(self._owner_task) if self._owner_task else None
1851          return LockStatistics(self.locked(), task_info, len(self._waiters))
1852  
1853  
1854  class Semaphore(BaseSemaphore):
1855      def __new__(
1856          cls,
1857          initial_value: int,
1858          *,
1859          max_value: int | None = None,
1860          fast_acquire: bool = False,
1861      ) -> Semaphore:
1862          return object.__new__(cls)
1863  
1864      def __init__(
1865          self,
1866          initial_value: int,
1867          *,
1868          max_value: int | None = None,
1869          fast_acquire: bool = False,
1870      ):
1871          super().__init__(initial_value, max_value=max_value)
1872          self._value = initial_value
1873          self._max_value = max_value
1874          self._fast_acquire = fast_acquire
1875          self._waiters: deque[asyncio.Future[None]] = deque()
1876  
1877      async def acquire(self) -> None:
1878          if self._value > 0 and not self._waiters:
1879              await AsyncIOBackend.checkpoint_if_cancelled()
1880              self._value -= 1
1881  
1882              # Unless on the "fast path", yield control of the event loop so that other
1883              # tasks can run too
1884              if not self._fast_acquire:
1885                  try:
1886                      await AsyncIOBackend.cancel_shielded_checkpoint()
1887                  except CancelledError:
1888                      self.release()
1889                      raise
1890  
1891              return
1892  
1893          fut: asyncio.Future[None] = asyncio.Future()
1894          self._waiters.append(fut)
1895          try:
1896              await fut
1897          except CancelledError:
1898              try:
1899                  self._waiters.remove(fut)
1900              except ValueError:
1901                  self.release()
1902  
1903              raise
1904  
1905      def acquire_nowait(self) -> None:
1906          if self._value == 0:
1907              raise WouldBlock
1908  
1909          self._value -= 1
1910  
1911      def release(self) -> None:
1912          if self._max_value is not None and self._value == self._max_value:
1913              raise ValueError("semaphore released too many times")
1914  
1915          for fut in self._waiters:
1916              if not fut.cancelled():
1917                  fut.set_result(None)
1918                  self._waiters.remove(fut)
1919                  return
1920  
1921          self._value += 1
1922  
1923      @property
1924      def value(self) -> int:
1925          return self._value
1926  
1927      @property
1928      def max_value(self) -> int | None:
1929          return self._max_value
1930  
1931      def statistics(self) -> SemaphoreStatistics:
1932          return SemaphoreStatistics(len(self._waiters))
1933  
1934  
1935  class CapacityLimiter(BaseCapacityLimiter):
1936      _total_tokens: float = 0
1937  
1938      def __new__(cls, total_tokens: float) -> CapacityLimiter:
1939          return object.__new__(cls)
1940  
1941      def __init__(self, total_tokens: float):
1942          self._borrowers: set[Any] = set()
1943          self._wait_queue: OrderedDict[Any, asyncio.Event] = OrderedDict()
1944          self.total_tokens = total_tokens
1945  
1946      async def __aenter__(self) -> None:
1947          await self.acquire()
1948  
1949      async def __aexit__(
1950          self,
1951          exc_type: type[BaseException] | None,
1952          exc_val: BaseException | None,
1953          exc_tb: TracebackType | None,
1954      ) -> None:
1955          self.release()
1956  
1957      @property
1958      def total_tokens(self) -> float:
1959          return self._total_tokens
1960  
1961      @total_tokens.setter
1962      def total_tokens(self, value: float) -> None:
1963          if not isinstance(value, int) and not math.isinf(value):
1964              raise TypeError("total_tokens must be an int or math.inf")
1965          if value < 1:
1966              raise ValueError("total_tokens must be >= 1")
1967  
1968          waiters_to_notify = max(value - self._total_tokens, 0)
1969          self._total_tokens = value
1970  
1971          # Notify waiting tasks that they have acquired the limiter
1972          while self._wait_queue and waiters_to_notify:
1973              event = self._wait_queue.popitem(last=False)[1]
1974              event.set()
1975              waiters_to_notify -= 1
1976  
1977      @property
1978      def borrowed_tokens(self) -> int:
1979          return len(self._borrowers)
1980  
1981      @property
1982      def available_tokens(self) -> float:
1983          return self._total_tokens - len(self._borrowers)
1984  
1985      def acquire_nowait(self) -> None:
1986          self.acquire_on_behalf_of_nowait(current_task())
1987  
1988      def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
1989          if borrower in self._borrowers:
1990              raise RuntimeError(
1991                  "this borrower is already holding one of this CapacityLimiter's tokens"
1992              )
1993  
1994          if self._wait_queue or len(self._borrowers) >= self._total_tokens:
1995              raise WouldBlock
1996  
1997          self._borrowers.add(borrower)
1998  
1999      async def acquire(self) -> None:
2000          return await self.acquire_on_behalf_of(current_task())
2001  
2002      async def acquire_on_behalf_of(self, borrower: object) -> None:
2003          await AsyncIOBackend.checkpoint_if_cancelled()
2004          try:
2005              self.acquire_on_behalf_of_nowait(borrower)
2006          except WouldBlock:
2007              event = asyncio.Event()
2008              self._wait_queue[borrower] = event
2009              try:
2010                  await event.wait()
2011              except BaseException:
2012                  self._wait_queue.pop(borrower, None)
2013                  raise
2014  
2015              self._borrowers.add(borrower)
2016          else:
2017              try:
2018                  await AsyncIOBackend.cancel_shielded_checkpoint()
2019              except BaseException:
2020                  self.release()
2021                  raise
2022  
2023      def release(self) -> None:
2024          self.release_on_behalf_of(current_task())
2025  
2026      def release_on_behalf_of(self, borrower: object) -> None:
2027          try:
2028              self._borrowers.remove(borrower)
2029          except KeyError:
2030              raise RuntimeError(
2031                  "this borrower isn't holding any of this CapacityLimiter's tokens"
2032              ) from None
2033  
2034          # Notify the next task in line if this limiter has free capacity now
2035          if self._wait_queue and len(self._borrowers) < self._total_tokens:
2036              event = self._wait_queue.popitem(last=False)[1]
2037              event.set()
2038  
2039      def statistics(self) -> CapacityLimiterStatistics:
2040          return CapacityLimiterStatistics(
2041              self.borrowed_tokens,
2042              self.total_tokens,
2043              tuple(self._borrowers),
2044              len(self._wait_queue),
2045          )
2046  
2047  
2048  _default_thread_limiter: RunVar[CapacityLimiter] = RunVar("_default_thread_limiter")
2049  
2050  
2051  #
2052  # Operating system signals
2053  #
2054  
2055  
2056  class _SignalReceiver:
2057      def __init__(self, signals: tuple[Signals, ...]):
2058          self._signals = signals
2059          self._loop = get_running_loop()
2060          self._signal_queue: deque[Signals] = deque()
2061          self._future: asyncio.Future = asyncio.Future()
2062          self._handled_signals: set[Signals] = set()
2063  
2064      def _deliver(self, signum: Signals) -> None:
2065          self._signal_queue.append(signum)
2066          if not self._future.done():
2067              self._future.set_result(None)
2068  
2069      def __enter__(self) -> _SignalReceiver:
2070          for sig in set(self._signals):
2071              self._loop.add_signal_handler(sig, self._deliver, sig)
2072              self._handled_signals.add(sig)
2073  
2074          return self
2075  
2076      def __exit__(
2077          self,
2078          exc_type: type[BaseException] | None,
2079          exc_val: BaseException | None,
2080          exc_tb: TracebackType | None,
2081      ) -> None:
2082          for sig in self._handled_signals:
2083              self._loop.remove_signal_handler(sig)
2084  
2085      def __aiter__(self) -> _SignalReceiver:
2086          return self
2087  
2088      async def __anext__(self) -> Signals:
2089          await AsyncIOBackend.checkpoint()
2090          if not self._signal_queue:
2091              self._future = asyncio.Future()
2092              await self._future
2093  
2094          return self._signal_queue.popleft()
2095  
2096  
2097  #
2098  # Testing and debugging
2099  #
2100  
2101  
2102  class AsyncIOTaskInfo(TaskInfo):
2103      def __init__(self, task: asyncio.Task):
2104          task_state = _task_states.get(task)
2105          if task_state is None:
2106              parent_id = None
2107          else:
2108              parent_id = task_state.parent_id
2109  
2110          coro = task.get_coro()
2111          assert coro is not None, "created TaskInfo from a completed Task"
2112          super().__init__(id(task), parent_id, task.get_name(), coro)
2113          self._task = weakref.ref(task)
2114  
2115      def has_pending_cancellation(self) -> bool:
2116          if not (task := self._task()):
2117              # If the task isn't around anymore, it won't have a pending cancellation
2118              return False
2119  
2120          if task._must_cancel:  # type: ignore[attr-defined]
2121              return True
2122          elif (
2123              isinstance(task._fut_waiter, asyncio.Future)  # type: ignore[attr-defined]
2124              and task._fut_waiter.cancelled()  # type: ignore[attr-defined]
2125          ):
2126              return True
2127  
2128          if task_state := _task_states.get(task):
2129              if cancel_scope := task_state.cancel_scope:
2130                  return cancel_scope._effectively_cancelled
2131  
2132          return False
2133  
2134  
2135  class TestRunner(abc.TestRunner):
2136      _send_stream: MemoryObjectSendStream[tuple[Awaitable[Any], asyncio.Future[Any]]]
2137  
2138      def __init__(
2139          self,
2140          *,
2141          debug: bool | None = None,
2142          use_uvloop: bool = False,
2143          loop_factory: Callable[[], AbstractEventLoop] | None = None,
2144      ) -> None:
2145          if use_uvloop and loop_factory is None:
2146              import uvloop
2147  
2148              loop_factory = uvloop.new_event_loop
2149  
2150          self._runner = Runner(debug=debug, loop_factory=loop_factory)
2151          self._exceptions: list[BaseException] = []
2152          self._runner_task: asyncio.Task | None = None
2153  
2154      def __enter__(self) -> TestRunner:
2155          self._runner.__enter__()
2156          self.get_loop().set_exception_handler(self._exception_handler)
2157          return self
2158  
2159      def __exit__(
2160          self,
2161          exc_type: type[BaseException] | None,
2162          exc_val: BaseException | None,
2163          exc_tb: TracebackType | None,
2164      ) -> None:
2165          self._runner.__exit__(exc_type, exc_val, exc_tb)
2166  
2167      def get_loop(self) -> AbstractEventLoop:
2168          return self._runner.get_loop()
2169  
2170      def _exception_handler(
2171          self, loop: asyncio.AbstractEventLoop, context: dict[str, Any]
2172      ) -> None:
2173          if isinstance(context.get("exception"), Exception):
2174              self._exceptions.append(context["exception"])
2175          else:
2176              loop.default_exception_handler(context)
2177  
2178      def _raise_async_exceptions(self) -> None:
2179          # Re-raise any exceptions raised in asynchronous callbacks
2180          if self._exceptions:
2181              exceptions, self._exceptions = self._exceptions, []
2182              if len(exceptions) == 1:
2183                  raise exceptions[0]
2184              elif exceptions:
2185                  raise BaseExceptionGroup(
2186                      "Multiple exceptions occurred in asynchronous callbacks", exceptions
2187                  )
2188  
2189      async def _run_tests_and_fixtures(
2190          self,
2191          receive_stream: MemoryObjectReceiveStream[
2192              tuple[Awaitable[T_Retval], asyncio.Future[T_Retval]]
2193          ],
2194      ) -> None:
2195          from _pytest.outcomes import OutcomeException
2196  
2197          with receive_stream, self._send_stream:
2198              async for coro, future in receive_stream:
2199                  try:
2200                      retval = await coro
2201                  except CancelledError as exc:
2202                      if not future.cancelled():
2203                          future.cancel(*exc.args)
2204  
2205                      raise
2206                  except BaseException as exc:
2207                      if not future.cancelled():
2208                          future.set_exception(exc)
2209  
2210                      if not isinstance(exc, (Exception, OutcomeException)):
2211                          raise
2212                  else:
2213                      if not future.cancelled():
2214                          future.set_result(retval)
2215  
2216      async def _call_in_runner_task(
2217          self,
2218          func: Callable[P, Awaitable[T_Retval]],
2219          *args: P.args,
2220          **kwargs: P.kwargs,
2221      ) -> T_Retval:
2222          if not self._runner_task:
2223              self._send_stream, receive_stream = create_memory_object_stream[
2224                  tuple[Awaitable[Any], asyncio.Future]
2225              ](1)
2226              self._runner_task = self.get_loop().create_task(
2227                  self._run_tests_and_fixtures(receive_stream)
2228              )
2229  
2230          coro = func(*args, **kwargs)
2231          future: asyncio.Future[T_Retval] = self.get_loop().create_future()
2232          self._send_stream.send_nowait((coro, future))
2233          return await future
2234  
2235      def run_asyncgen_fixture(
2236          self,
2237          fixture_func: Callable[..., AsyncGenerator[T_Retval, Any]],
2238          kwargs: dict[str, Any],
2239      ) -> Iterable[T_Retval]:
2240          asyncgen = fixture_func(**kwargs)
2241          fixturevalue: T_Retval = self.get_loop().run_until_complete(
2242              self._call_in_runner_task(asyncgen.asend, None)
2243          )
2244          self._raise_async_exceptions()
2245  
2246          yield fixturevalue
2247  
2248          try:
2249              self.get_loop().run_until_complete(
2250                  self._call_in_runner_task(asyncgen.asend, None)
2251              )
2252          except StopAsyncIteration:
2253              self._raise_async_exceptions()
2254          else:
2255              self.get_loop().run_until_complete(asyncgen.aclose())
2256              raise RuntimeError("Async generator fixture did not stop")
2257  
2258      def run_fixture(
2259          self,
2260          fixture_func: Callable[..., Coroutine[Any, Any, T_Retval]],
2261          kwargs: dict[str, Any],
2262      ) -> T_Retval:
2263          retval = self.get_loop().run_until_complete(
2264              self._call_in_runner_task(fixture_func, **kwargs)
2265          )
2266          self._raise_async_exceptions()
2267          return retval
2268  
2269      def run_test(
2270          self, test_func: Callable[..., Coroutine[Any, Any, Any]], kwargs: dict[str, Any]
2271      ) -> None:
2272          try:
2273              self.get_loop().run_until_complete(
2274                  self._call_in_runner_task(test_func, **kwargs)
2275              )
2276          except Exception as exc:
2277              self._exceptions.append(exc)
2278  
2279          self._raise_async_exceptions()
2280  
2281  
2282  class AsyncIOBackend(AsyncBackend):
2283      @classmethod
2284      def run(
2285          cls,
2286          func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
2287          args: tuple[Unpack[PosArgsT]],
2288          kwargs: dict[str, Any],
2289          options: dict[str, Any],
2290      ) -> T_Retval:
2291          @wraps(func)
2292          async def wrapper() -> T_Retval:
2293              task = cast(asyncio.Task, current_task())
2294              task.set_name(get_callable_name(func))
2295              _task_states[task] = TaskState(None, None)
2296  
2297              try:
2298                  return await func(*args)
2299              finally:
2300                  del _task_states[task]
2301  
2302          debug = options.get("debug", None)
2303          loop_factory = options.get("loop_factory", None)
2304          if loop_factory is None and options.get("use_uvloop", False):
2305              import uvloop
2306  
2307              loop_factory = uvloop.new_event_loop
2308  
2309          with Runner(debug=debug, loop_factory=loop_factory) as runner:
2310              return runner.run(wrapper())
2311  
2312      @classmethod
2313      def current_token(cls) -> object:
2314          return get_running_loop()
2315  
2316      @classmethod
2317      def current_time(cls) -> float:
2318          return get_running_loop().time()
2319  
2320      @classmethod
2321      def cancelled_exception_class(cls) -> type[BaseException]:
2322          return CancelledError
2323  
2324      @classmethod
2325      async def checkpoint(cls) -> None:
2326          await sleep(0)
2327  
2328      @classmethod
2329      async def checkpoint_if_cancelled(cls) -> None:
2330          task = current_task()
2331          if task is None:
2332              return
2333  
2334          try:
2335              cancel_scope = _task_states[task].cancel_scope
2336          except KeyError:
2337              return
2338  
2339          while cancel_scope:
2340              if cancel_scope.cancel_called:
2341                  await sleep(0)
2342              elif cancel_scope.shield:
2343                  break
2344              else:
2345                  cancel_scope = cancel_scope._parent_scope
2346  
2347      @classmethod
2348      async def cancel_shielded_checkpoint(cls) -> None:
2349          with CancelScope(shield=True):
2350              await sleep(0)
2351  
2352      @classmethod
2353      async def sleep(cls, delay: float) -> None:
2354          await sleep(delay)
2355  
2356      @classmethod
2357      def create_cancel_scope(
2358          cls, *, deadline: float = math.inf, shield: bool = False
2359      ) -> CancelScope:
2360          return CancelScope(deadline=deadline, shield=shield)
2361  
2362      @classmethod
2363      def current_effective_deadline(cls) -> float:
2364          if (task := current_task()) is None:
2365              return math.inf
2366  
2367          try:
2368              cancel_scope = _task_states[task].cancel_scope
2369          except KeyError:
2370              return math.inf
2371  
2372          deadline = math.inf
2373          while cancel_scope:
2374              deadline = min(deadline, cancel_scope.deadline)
2375              if cancel_scope._cancel_called:
2376                  deadline = -math.inf
2377                  break
2378              elif cancel_scope.shield:
2379                  break
2380              else:
2381                  cancel_scope = cancel_scope._parent_scope
2382  
2383          return deadline
2384  
2385      @classmethod
2386      def create_task_group(cls) -> abc.TaskGroup:
2387          return TaskGroup()
2388  
2389      @classmethod
2390      def create_event(cls) -> abc.Event:
2391          return Event()
2392  
2393      @classmethod
2394      def create_lock(cls, *, fast_acquire: bool) -> abc.Lock:
2395          return Lock(fast_acquire=fast_acquire)
2396  
2397      @classmethod
2398      def create_semaphore(
2399          cls,
2400          initial_value: int,
2401          *,
2402          max_value: int | None = None,
2403          fast_acquire: bool = False,
2404      ) -> abc.Semaphore:
2405          return Semaphore(initial_value, max_value=max_value, fast_acquire=fast_acquire)
2406  
2407      @classmethod
2408      def create_capacity_limiter(cls, total_tokens: float) -> abc.CapacityLimiter:
2409          return CapacityLimiter(total_tokens)
2410  
2411      @classmethod
2412      async def run_sync_in_worker_thread(  # type: ignore[return]
2413          cls,
2414          func: Callable[[Unpack[PosArgsT]], T_Retval],
2415          args: tuple[Unpack[PosArgsT]],
2416          abandon_on_cancel: bool = False,
2417          limiter: abc.CapacityLimiter | None = None,
2418      ) -> T_Retval:
2419          await cls.checkpoint()
2420  
2421          # If this is the first run in this event loop thread, set up the necessary
2422          # variables
2423          try:
2424              idle_workers = _threadpool_idle_workers.get()
2425              workers = _threadpool_workers.get()
2426          except LookupError:
2427              idle_workers = deque()
2428              workers = set()
2429              _threadpool_idle_workers.set(idle_workers)
2430              _threadpool_workers.set(workers)
2431  
2432          async with limiter or cls.current_default_thread_limiter():
2433              with CancelScope(shield=not abandon_on_cancel) as scope:
2434                  future = asyncio.Future[T_Retval]()
2435                  root_task = find_root_task()
2436                  if not idle_workers:
2437                      worker = WorkerThread(root_task, workers, idle_workers)
2438                      worker.start()
2439                      workers.add(worker)
2440                      root_task.add_done_callback(
2441                          worker.stop, context=contextvars.Context()
2442                      )
2443                  else:
2444                      worker = idle_workers.pop()
2445  
2446                      # Prune any other workers that have been idle for MAX_IDLE_TIME
2447                      # seconds or longer
2448                      now = cls.current_time()
2449                      while idle_workers:
2450                          if (
2451                              now - idle_workers[0].idle_since
2452                              < WorkerThread.MAX_IDLE_TIME
2453                          ):
2454                              break
2455  
2456                          expired_worker = idle_workers.popleft()
2457                          expired_worker.root_task.remove_done_callback(
2458                              expired_worker.stop
2459                          )
2460                          expired_worker.stop()
2461  
2462                  context = copy_context()
2463                  context.run(sniffio.current_async_library_cvar.set, None)
2464                  if abandon_on_cancel or scope._parent_scope is None:
2465                      worker_scope = scope
2466                  else:
2467                      worker_scope = scope._parent_scope
2468  
2469                  worker.queue.put_nowait((context, func, args, future, worker_scope))
2470                  return await future
2471  
2472      @classmethod
2473      def check_cancelled(cls) -> None:
2474          scope: CancelScope | None = threadlocals.current_cancel_scope
2475          while scope is not None:
2476              if scope.cancel_called:
2477                  raise CancelledError(f"Cancelled by cancel scope {id(scope):x}")
2478  
2479              if scope.shield:
2480                  return
2481  
2482              scope = scope._parent_scope
2483  
2484      @classmethod
2485      def run_async_from_thread(
2486          cls,
2487          func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
2488          args: tuple[Unpack[PosArgsT]],
2489          token: object,
2490      ) -> T_Retval:
2491          async def task_wrapper(scope: CancelScope) -> T_Retval:
2492              __tracebackhide__ = True
2493              task = cast(asyncio.Task, current_task())
2494              _task_states[task] = TaskState(None, scope)
2495              scope._tasks.add(task)
2496              try:
2497                  return await func(*args)
2498              except CancelledError as exc:
2499                  raise concurrent.futures.CancelledError(str(exc)) from None
2500              finally:
2501                  scope._tasks.discard(task)
2502  
2503          loop = cast(AbstractEventLoop, token)
2504          context = copy_context()
2505          context.run(sniffio.current_async_library_cvar.set, "asyncio")
2506          wrapper = task_wrapper(threadlocals.current_cancel_scope)
2507          f: concurrent.futures.Future[T_Retval] = context.run(
2508              asyncio.run_coroutine_threadsafe, wrapper, loop
2509          )
2510          return f.result()
2511  
2512      @classmethod
2513      def run_sync_from_thread(
2514          cls,
2515          func: Callable[[Unpack[PosArgsT]], T_Retval],
2516          args: tuple[Unpack[PosArgsT]],
2517          token: object,
2518      ) -> T_Retval:
2519          @wraps(func)
2520          def wrapper() -> None:
2521              try:
2522                  sniffio.current_async_library_cvar.set("asyncio")
2523                  f.set_result(func(*args))
2524              except BaseException as exc:
2525                  f.set_exception(exc)
2526                  if not isinstance(exc, Exception):
2527                      raise
2528  
2529          f: concurrent.futures.Future[T_Retval] = Future()
2530          loop = cast(AbstractEventLoop, token)
2531          loop.call_soon_threadsafe(wrapper)
2532          return f.result()
2533  
2534      @classmethod
2535      def create_blocking_portal(cls) -> abc.BlockingPortal:
2536          return BlockingPortal()
2537  
2538      @classmethod
2539      async def open_process(
2540          cls,
2541          command: StrOrBytesPath | Sequence[StrOrBytesPath],
2542          *,
2543          stdin: int | IO[Any] | None,
2544          stdout: int | IO[Any] | None,
2545          stderr: int | IO[Any] | None,
2546          **kwargs: Any,
2547      ) -> Process:
2548          await cls.checkpoint()
2549          if isinstance(command, PathLike):
2550              command = os.fspath(command)
2551  
2552          if isinstance(command, (str, bytes)):
2553              process = await asyncio.create_subprocess_shell(
2554                  command,
2555                  stdin=stdin,
2556                  stdout=stdout,
2557                  stderr=stderr,
2558                  **kwargs,
2559              )
2560          else:
2561              process = await asyncio.create_subprocess_exec(
2562                  *command,
2563                  stdin=stdin,
2564                  stdout=stdout,
2565                  stderr=stderr,
2566                  **kwargs,
2567              )
2568  
2569          stdin_stream = StreamWriterWrapper(process.stdin) if process.stdin else None
2570          stdout_stream = StreamReaderWrapper(process.stdout) if process.stdout else None
2571          stderr_stream = StreamReaderWrapper(process.stderr) if process.stderr else None
2572          return Process(process, stdin_stream, stdout_stream, stderr_stream)
2573  
2574      @classmethod
2575      def setup_process_pool_exit_at_shutdown(cls, workers: set[abc.Process]) -> None:
2576          create_task(
2577              _shutdown_process_pool_on_exit(workers),
2578              name="AnyIO process pool shutdown task",
2579          )
2580          find_root_task().add_done_callback(
2581              partial(_forcibly_shutdown_process_pool_on_exit, workers)  # type:ignore[arg-type]
2582          )
2583  
2584      @classmethod
2585      async def connect_tcp(
2586          cls, host: str, port: int, local_address: IPSockAddrType | None = None
2587      ) -> abc.SocketStream:
2588          transport, protocol = cast(
2589              tuple[asyncio.Transport, StreamProtocol],
2590              await get_running_loop().create_connection(
2591                  StreamProtocol, host, port, local_addr=local_address
2592              ),
2593          )
2594          transport.pause_reading()
2595          return SocketStream(transport, protocol)
2596  
2597      @classmethod
2598      async def connect_unix(cls, path: str | bytes) -> abc.UNIXSocketStream:
2599          await cls.checkpoint()
2600          loop = get_running_loop()
2601          raw_socket = socket.socket(socket.AF_UNIX)
2602          raw_socket.setblocking(False)
2603          while True:
2604              try:
2605                  raw_socket.connect(path)
2606              except BlockingIOError:
2607                  f: asyncio.Future = asyncio.Future()
2608                  loop.add_writer(raw_socket, f.set_result, None)
2609                  f.add_done_callback(lambda _: loop.remove_writer(raw_socket))
2610                  await f
2611              except BaseException:
2612                  raw_socket.close()
2613                  raise
2614              else:
2615                  return UNIXSocketStream(raw_socket)
2616  
2617      @classmethod
2618      def create_tcp_listener(cls, sock: socket.socket) -> SocketListener:
2619          return TCPSocketListener(sock)
2620  
2621      @classmethod
2622      def create_unix_listener(cls, sock: socket.socket) -> SocketListener:
2623          return UNIXSocketListener(sock)
2624  
2625      @classmethod
2626      async def create_udp_socket(
2627          cls,
2628          family: AddressFamily,
2629          local_address: IPSockAddrType | None,
2630          remote_address: IPSockAddrType | None,
2631          reuse_port: bool,
2632      ) -> UDPSocket | ConnectedUDPSocket:
2633          transport, protocol = await get_running_loop().create_datagram_endpoint(
2634              DatagramProtocol,
2635              local_addr=local_address,
2636              remote_addr=remote_address,
2637              family=family,
2638              reuse_port=reuse_port,
2639          )
2640          if protocol.exception:
2641              transport.close()
2642              raise protocol.exception
2643  
2644          if not remote_address:
2645              return UDPSocket(transport, protocol)
2646          else:
2647              return ConnectedUDPSocket(transport, protocol)
2648  
2649      @classmethod
2650      async def create_unix_datagram_socket(  # type: ignore[override]
2651          cls, raw_socket: socket.socket, remote_path: str | bytes | None
2652      ) -> abc.UNIXDatagramSocket | abc.ConnectedUNIXDatagramSocket:
2653          await cls.checkpoint()
2654          loop = get_running_loop()
2655  
2656          if remote_path:
2657              while True:
2658                  try:
2659                      raw_socket.connect(remote_path)
2660                  except BlockingIOError:
2661                      f: asyncio.Future = asyncio.Future()
2662                      loop.add_writer(raw_socket, f.set_result, None)
2663                      f.add_done_callback(lambda _: loop.remove_writer(raw_socket))
2664                      await f
2665                  except BaseException:
2666                      raw_socket.close()
2667                      raise
2668                  else:
2669                      return ConnectedUNIXDatagramSocket(raw_socket)
2670          else:
2671              return UNIXDatagramSocket(raw_socket)
2672  
2673      @classmethod
2674      async def getaddrinfo(
2675          cls,
2676          host: bytes | str | None,
2677          port: str | int | None,
2678          *,
2679          family: int | AddressFamily = 0,
2680          type: int | SocketKind = 0,
2681          proto: int = 0,
2682          flags: int = 0,
2683      ) -> Sequence[
2684          tuple[
2685              AddressFamily,
2686              SocketKind,
2687              int,
2688              str,
2689              tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes],
2690          ]
2691      ]:
2692          return await get_running_loop().getaddrinfo(
2693              host, port, family=family, type=type, proto=proto, flags=flags
2694          )
2695  
2696      @classmethod
2697      async def getnameinfo(
2698          cls, sockaddr: IPSockAddrType, flags: int = 0
2699      ) -> tuple[str, str]:
2700          return await get_running_loop().getnameinfo(sockaddr, flags)
2701  
2702      @classmethod
2703      async def wait_readable(cls, obj: FileDescriptorLike) -> None:
2704          await cls.checkpoint()
2705          try:
2706              read_events = _read_events.get()
2707          except LookupError:
2708              read_events = {}
2709              _read_events.set(read_events)
2710  
2711          if not isinstance(obj, int):
2712              obj = obj.fileno()
2713  
2714          if read_events.get(obj):
2715              raise BusyResourceError("reading from")
2716  
2717          loop = get_running_loop()
2718          event = asyncio.Event()
2719          try:
2720              loop.add_reader(obj, event.set)
2721          except NotImplementedError:
2722              from anyio._core._asyncio_selector_thread import get_selector
2723  
2724              selector = get_selector()
2725              selector.add_reader(obj, event.set)
2726              remove_reader = selector.remove_reader
2727          else:
2728              remove_reader = loop.remove_reader
2729  
2730          read_events[obj] = event
2731          try:
2732              await event.wait()
2733          finally:
2734              remove_reader(obj)
2735              del read_events[obj]
2736  
2737      @classmethod
2738      async def wait_writable(cls, obj: FileDescriptorLike) -> None:
2739          await cls.checkpoint()
2740          try:
2741              write_events = _write_events.get()
2742          except LookupError:
2743              write_events = {}
2744              _write_events.set(write_events)
2745  
2746          if not isinstance(obj, int):
2747              obj = obj.fileno()
2748  
2749          if write_events.get(obj):
2750              raise BusyResourceError("writing to")
2751  
2752          loop = get_running_loop()
2753          event = asyncio.Event()
2754          try:
2755              loop.add_writer(obj, event.set)
2756          except NotImplementedError:
2757              from anyio._core._asyncio_selector_thread import get_selector
2758  
2759              selector = get_selector()
2760              selector.add_writer(obj, event.set)
2761              remove_writer = selector.remove_writer
2762          else:
2763              remove_writer = loop.remove_writer
2764  
2765          write_events[obj] = event
2766          try:
2767              await event.wait()
2768          finally:
2769              del write_events[obj]
2770              remove_writer(obj)
2771  
2772      @classmethod
2773      def current_default_thread_limiter(cls) -> CapacityLimiter:
2774          try:
2775              return _default_thread_limiter.get()
2776          except LookupError:
2777              limiter = CapacityLimiter(40)
2778              _default_thread_limiter.set(limiter)
2779              return limiter
2780  
2781      @classmethod
2782      def open_signal_receiver(
2783          cls, *signals: Signals
2784      ) -> AbstractContextManager[AsyncIterator[Signals]]:
2785          return _SignalReceiver(signals)
2786  
2787      @classmethod
2788      def get_current_task(cls) -> TaskInfo:
2789          return AsyncIOTaskInfo(current_task())  # type: ignore[arg-type]
2790  
2791      @classmethod
2792      def get_running_tasks(cls) -> Sequence[TaskInfo]:
2793          return [AsyncIOTaskInfo(task) for task in all_tasks() if not task.done()]
2794  
2795      @classmethod
2796      async def wait_all_tasks_blocked(cls) -> None:
2797          await cls.checkpoint()
2798          this_task = current_task()
2799          while True:
2800              for task in all_tasks():
2801                  if task is this_task:
2802                      continue
2803  
2804                  waiter = task._fut_waiter  # type: ignore[attr-defined]
2805                  if waiter is None or waiter.done():
2806                      await sleep(0.1)
2807                      break
2808              else:
2809                  return
2810  
2811      @classmethod
2812      def create_test_runner(cls, options: dict[str, Any]) -> TestRunner:
2813          return TestRunner(**options)
2814  
2815  
2816  backend_class = AsyncIOBackend
2817