Загрузить файлы в «venv/Lib/site-packages/anyio»
This commit is contained in:
BIN
venv/Lib/site-packages/anyio/py.typed
Normal file
BIN
venv/Lib/site-packages/anyio/py.typed
Normal file
Binary file not shown.
375
venv/Lib/site-packages/anyio/pytest_plugin.py
Normal file
375
venv/Lib/site-packages/anyio/pytest_plugin.py
Normal file
@@ -0,0 +1,375 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import socket
|
||||
import sys
|
||||
from collections.abc import Callable, Generator, Iterator
|
||||
from contextlib import ExitStack, contextmanager
|
||||
from inspect import isasyncgenfunction, iscoroutinefunction, ismethod
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
from _pytest.fixtures import FuncFixtureInfo, SubRequest
|
||||
from _pytest.outcomes import Exit
|
||||
from _pytest.python import CallSpec2
|
||||
from _pytest.scope import Scope
|
||||
|
||||
from . import get_available_backends
|
||||
from ._core._eventloop import (
|
||||
current_async_library,
|
||||
get_async_backend,
|
||||
reset_current_async_library,
|
||||
set_current_async_library,
|
||||
)
|
||||
from ._core._exceptions import iterate_exceptions
|
||||
from .abc import TestRunner
|
||||
|
||||
if sys.version_info < (3, 11):
|
||||
from exceptiongroup import ExceptionGroup
|
||||
|
||||
_current_runner: TestRunner | None = None
|
||||
_runner_stack: ExitStack | None = None
|
||||
_runner_leases = 0
|
||||
|
||||
|
||||
def extract_backend_and_options(backend: object) -> tuple[str, dict[str, Any]]:
|
||||
if isinstance(backend, str):
|
||||
return backend, {}
|
||||
elif isinstance(backend, tuple) and len(backend) == 2:
|
||||
if isinstance(backend[0], str) and isinstance(backend[1], dict):
|
||||
return cast(tuple[str, dict[str, Any]], backend)
|
||||
|
||||
raise TypeError("anyio_backend must be either a string or tuple of (string, dict)")
|
||||
|
||||
|
||||
@contextmanager
|
||||
def get_runner(
|
||||
backend_name: str, backend_options: dict[str, Any]
|
||||
) -> Iterator[TestRunner]:
|
||||
global _current_runner, _runner_leases, _runner_stack
|
||||
if _current_runner is None:
|
||||
asynclib = get_async_backend(backend_name)
|
||||
_runner_stack = ExitStack()
|
||||
if current_async_library() is None:
|
||||
# Since we're in control of the event loop, we can cache the name of the
|
||||
# async library
|
||||
token = set_current_async_library(backend_name)
|
||||
_runner_stack.callback(reset_current_async_library, token)
|
||||
|
||||
backend_options = backend_options or {}
|
||||
_current_runner = _runner_stack.enter_context(
|
||||
asynclib.create_test_runner(backend_options)
|
||||
)
|
||||
|
||||
_runner_leases += 1
|
||||
try:
|
||||
yield _current_runner
|
||||
finally:
|
||||
_runner_leases -= 1
|
||||
if not _runner_leases:
|
||||
assert _runner_stack is not None
|
||||
_runner_stack.close()
|
||||
_runner_stack = _current_runner = None
|
||||
|
||||
|
||||
def pytest_addoption(parser: pytest.Parser) -> None:
|
||||
parser.addini(
|
||||
"anyio_mode",
|
||||
default="strict",
|
||||
help='AnyIO plugin mode (either "strict" or "auto")',
|
||||
)
|
||||
|
||||
|
||||
def pytest_configure(config: pytest.Config) -> None:
|
||||
config.addinivalue_line(
|
||||
"markers",
|
||||
"anyio: mark the (coroutine function) test to be run asynchronously via anyio.",
|
||||
)
|
||||
if (
|
||||
config.getini("anyio_mode") == "auto"
|
||||
and config.pluginmanager.has_plugin("asyncio")
|
||||
and config.getini("asyncio_mode") == "auto"
|
||||
):
|
||||
config.issue_config_time_warning(
|
||||
pytest.PytestConfigWarning(
|
||||
"AnyIO auto mode has been enabled together with pytest-asyncio auto "
|
||||
"mode. This may cause unexpected behavior."
|
||||
),
|
||||
1,
|
||||
)
|
||||
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_fixture_setup(fixturedef: Any, request: Any) -> Generator[Any]:
|
||||
def wrapper(anyio_backend: Any, request: SubRequest, **kwargs: Any) -> Any:
|
||||
# Rebind any fixture methods to the request instance
|
||||
if (
|
||||
request.instance
|
||||
and ismethod(func)
|
||||
and type(func.__self__) is type(request.instance)
|
||||
):
|
||||
local_func = func.__func__.__get__(request.instance)
|
||||
else:
|
||||
local_func = func
|
||||
|
||||
backend_name, backend_options = extract_backend_and_options(anyio_backend)
|
||||
if has_backend_arg:
|
||||
kwargs["anyio_backend"] = anyio_backend
|
||||
|
||||
if has_request_arg:
|
||||
kwargs["request"] = request
|
||||
|
||||
with get_runner(backend_name, backend_options) as runner:
|
||||
# re-entrant call into the test runner detected. this happens when an async fixture
|
||||
# is dynamically requested via request.getfixturevalue() from inside a running async
|
||||
# test or fixture. on asyncio this raises RuntimeError: This event loop is already
|
||||
# running, on trio the runner deadlocks - the host loop blocks waiting for the
|
||||
# coroutine to return, but the coroutine is waiting for the host loop. raising here
|
||||
# prevents the hang and gives a consistent error across backends.
|
||||
if runner.is_running():
|
||||
raise RuntimeError(
|
||||
"Cannot schedule a coroutine in the test runner while another is already running; "
|
||||
"likely caused by request.getfixturevalue() on an async fixture."
|
||||
)
|
||||
|
||||
if isasyncgenfunction(local_func):
|
||||
yield from runner.run_asyncgen_fixture(local_func, kwargs)
|
||||
else:
|
||||
yield runner.run_fixture(local_func, kwargs)
|
||||
|
||||
# Only apply this to coroutine functions and async generator functions in requests
|
||||
# that involve the anyio_backend fixture
|
||||
func = fixturedef.func
|
||||
if isasyncgenfunction(func) or iscoroutinefunction(func):
|
||||
if "anyio_backend" in request.fixturenames:
|
||||
fixturedef.func = wrapper
|
||||
original_argname = fixturedef.argnames
|
||||
|
||||
if not (has_backend_arg := "anyio_backend" in fixturedef.argnames):
|
||||
fixturedef.argnames += ("anyio_backend",)
|
||||
|
||||
if not (has_request_arg := "request" in fixturedef.argnames):
|
||||
fixturedef.argnames += ("request",)
|
||||
|
||||
try:
|
||||
return (yield)
|
||||
finally:
|
||||
fixturedef.func = func
|
||||
fixturedef.argnames = original_argname
|
||||
|
||||
return (yield)
|
||||
|
||||
|
||||
@pytest.hookimpl(tryfirst=True)
|
||||
def pytest_pycollect_makeitem(
|
||||
collector: pytest.Module | pytest.Class, name: str, obj: object
|
||||
) -> None:
|
||||
if collector.istestfunction(obj, name):
|
||||
inner_func = obj.hypothesis.inner_test if hasattr(obj, "hypothesis") else obj
|
||||
if iscoroutinefunction(inner_func):
|
||||
anyio_auto_mode = collector.config.getini("anyio_mode") == "auto"
|
||||
marker = collector.get_closest_marker("anyio")
|
||||
own_markers = getattr(obj, "pytestmark", ())
|
||||
if (
|
||||
anyio_auto_mode
|
||||
or marker
|
||||
or any(marker.name == "anyio" for marker in own_markers)
|
||||
):
|
||||
pytest.mark.usefixtures("anyio_backend")(obj)
|
||||
|
||||
|
||||
def pytest_collection_finish(session: pytest.Session) -> None:
|
||||
for i, item in reversed(list(enumerate(session.items))):
|
||||
if (
|
||||
isinstance(item, pytest.Function)
|
||||
and iscoroutinefunction(item.function)
|
||||
and item.get_closest_marker("anyio") is not None
|
||||
and "anyio_backend" not in item.fixturenames
|
||||
):
|
||||
new_items = []
|
||||
try:
|
||||
cs_fields = {f.name for f in dataclasses.fields(CallSpec2)}
|
||||
except TypeError:
|
||||
cs_fields = set()
|
||||
|
||||
for param_index, backend in enumerate(get_available_backends()):
|
||||
if "_arg2scope" in cs_fields: # pytest >= 8
|
||||
callspec = CallSpec2(
|
||||
params={"anyio_backend": backend},
|
||||
indices={"anyio_backend": param_index},
|
||||
_arg2scope={"anyio_backend": Scope.Module},
|
||||
_idlist=[backend],
|
||||
marks=[],
|
||||
)
|
||||
else: # pytest 7.x
|
||||
callspec = CallSpec2( # type: ignore[call-arg]
|
||||
funcargs={},
|
||||
params={"anyio_backend": backend},
|
||||
indices={"anyio_backend": param_index},
|
||||
arg2scope={"anyio_backend": Scope.Module},
|
||||
idlist=[backend],
|
||||
marks=[],
|
||||
)
|
||||
|
||||
fi = item._fixtureinfo
|
||||
new_names_closure = list(fi.names_closure)
|
||||
if "anyio_backend" not in new_names_closure:
|
||||
new_names_closure.append("anyio_backend")
|
||||
|
||||
new_fixtureinfo = FuncFixtureInfo(
|
||||
argnames=fi.argnames,
|
||||
initialnames=fi.initialnames,
|
||||
names_closure=new_names_closure,
|
||||
name2fixturedefs=fi.name2fixturedefs,
|
||||
)
|
||||
new_item = pytest.Function.from_parent(
|
||||
item.parent,
|
||||
name=f"{item.originalname}[{backend}]",
|
||||
callspec=callspec,
|
||||
callobj=item.obj,
|
||||
fixtureinfo=new_fixtureinfo,
|
||||
keywords=item.keywords,
|
||||
originalname=item.originalname,
|
||||
)
|
||||
new_items.append(new_item)
|
||||
|
||||
session.items[i : i + 1] = new_items
|
||||
|
||||
|
||||
@pytest.hookimpl(tryfirst=True)
|
||||
def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
|
||||
def run_with_hypothesis(**kwargs: Any) -> None:
|
||||
with get_runner(backend_name, backend_options) as runner:
|
||||
runner.run_test(original_func, kwargs)
|
||||
|
||||
backend = pyfuncitem.funcargs.get("anyio_backend")
|
||||
if backend:
|
||||
backend_name, backend_options = extract_backend_and_options(backend)
|
||||
|
||||
if hasattr(pyfuncitem.obj, "hypothesis"):
|
||||
# Wrap the inner test function unless it's already wrapped
|
||||
original_func = pyfuncitem.obj.hypothesis.inner_test
|
||||
if original_func.__qualname__ != run_with_hypothesis.__qualname__:
|
||||
if iscoroutinefunction(original_func):
|
||||
pyfuncitem.obj.hypothesis.inner_test = run_with_hypothesis
|
||||
|
||||
return None
|
||||
|
||||
if iscoroutinefunction(pyfuncitem.obj):
|
||||
funcargs = pyfuncitem.funcargs
|
||||
testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
|
||||
with get_runner(backend_name, backend_options) as runner:
|
||||
try:
|
||||
runner.run_test(pyfuncitem.obj, testargs)
|
||||
except ExceptionGroup as excgrp:
|
||||
for exc in iterate_exceptions(excgrp):
|
||||
if isinstance(exc, (Exit, KeyboardInterrupt, SystemExit)):
|
||||
raise exc from excgrp
|
||||
|
||||
raise
|
||||
|
||||
return True
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", params=get_available_backends())
|
||||
def anyio_backend(request: Any) -> Any:
|
||||
return request.param
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend_name(anyio_backend: Any) -> str:
|
||||
if isinstance(anyio_backend, str):
|
||||
return anyio_backend
|
||||
else:
|
||||
return anyio_backend[0]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend_options(anyio_backend: Any) -> dict[str, Any]:
|
||||
if isinstance(anyio_backend, str):
|
||||
return {}
|
||||
else:
|
||||
return anyio_backend[1]
|
||||
|
||||
|
||||
class FreePortFactory:
|
||||
"""
|
||||
Manages port generation based on specified socket kind, ensuring no duplicate
|
||||
ports are generated.
|
||||
|
||||
This class provides functionality for generating available free ports on the
|
||||
system. It is initialized with a specific socket kind and can generate ports
|
||||
for given address families while avoiding reuse of previously generated ports.
|
||||
|
||||
Users should not instantiate this class directly, but use the
|
||||
``free_tcp_port_factory`` and ``free_udp_port_factory`` fixtures instead. For simple
|
||||
uses cases, ``free_tcp_port`` and ``free_udp_port`` can be used instead.
|
||||
"""
|
||||
|
||||
def __init__(self, kind: socket.SocketKind) -> None:
|
||||
self._kind = kind
|
||||
self._generated = set[int]()
|
||||
|
||||
@property
|
||||
def kind(self) -> socket.SocketKind:
|
||||
"""
|
||||
The type of socket connection (e.g., :data:`~socket.SOCK_STREAM` or
|
||||
:data:`~socket.SOCK_DGRAM`) used to bind for checking port availability
|
||||
|
||||
"""
|
||||
return self._kind
|
||||
|
||||
def __call__(self, family: socket.AddressFamily | None = None) -> int:
|
||||
"""
|
||||
Return an unbound port for the given address family.
|
||||
|
||||
:param family: if omitted, both IPv4 and IPv6 addresses will be tried
|
||||
:return: a port number
|
||||
|
||||
"""
|
||||
if family is not None:
|
||||
families = [family]
|
||||
else:
|
||||
families = [socket.AF_INET]
|
||||
if socket.has_ipv6:
|
||||
families.append(socket.AF_INET6)
|
||||
|
||||
while True:
|
||||
port = 0
|
||||
with ExitStack() as stack:
|
||||
for family in families:
|
||||
sock = stack.enter_context(socket.socket(family, self._kind))
|
||||
addr = "::1" if family == socket.AF_INET6 else "127.0.0.1"
|
||||
try:
|
||||
sock.bind((addr, port))
|
||||
except OSError:
|
||||
break
|
||||
|
||||
if not port:
|
||||
port = sock.getsockname()[1]
|
||||
else:
|
||||
if port not in self._generated:
|
||||
self._generated.add(port)
|
||||
return port
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def free_tcp_port_factory() -> FreePortFactory:
|
||||
return FreePortFactory(socket.SOCK_STREAM)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def free_udp_port_factory() -> FreePortFactory:
|
||||
return FreePortFactory(socket.SOCK_DGRAM)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def free_tcp_port(free_tcp_port_factory: Callable[[], int]) -> int:
|
||||
return free_tcp_port_factory()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def free_udp_port(free_udp_port_factory: Callable[[], int]) -> int:
|
||||
return free_udp_port_factory()
|
||||
246
venv/Lib/site-packages/anyio/to_interpreter.py
Normal file
246
venv/Lib/site-packages/anyio/to_interpreter.py
Normal file
@@ -0,0 +1,246 @@
|
||||
from __future__ import annotations
|
||||
|
||||
__all__ = (
|
||||
"run_sync",
|
||||
"current_default_interpreter_limiter",
|
||||
)
|
||||
|
||||
import atexit
|
||||
import os
|
||||
import sys
|
||||
from collections import deque
|
||||
from collections.abc import Callable
|
||||
from typing import Any, Final, TypeVar
|
||||
|
||||
from . import current_time, to_thread
|
||||
from ._core._exceptions import BrokenWorkerInterpreter
|
||||
from ._core._synchronization import CapacityLimiter
|
||||
from .lowlevel import RunVar
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from typing import TypeVarTuple, Unpack
|
||||
else:
|
||||
from typing_extensions import TypeVarTuple, Unpack
|
||||
|
||||
if sys.version_info >= (3, 14):
|
||||
from concurrent.interpreters import ExecutionFailed, create
|
||||
|
||||
def _interp_call(
|
||||
func: Callable[..., Any], args: tuple[Any, ...]
|
||||
) -> tuple[Any, bool]:
|
||||
try:
|
||||
retval = func(*args)
|
||||
except BaseException as exc:
|
||||
return exc, True
|
||||
else:
|
||||
return retval, False
|
||||
|
||||
class _Worker:
|
||||
last_used: float = 0
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._interpreter = create()
|
||||
|
||||
def destroy(self) -> None:
|
||||
self._interpreter.close()
|
||||
|
||||
def call(
|
||||
self,
|
||||
func: Callable[..., T_Retval],
|
||||
args: tuple[Any, ...],
|
||||
) -> T_Retval:
|
||||
try:
|
||||
res, is_exception = self._interpreter.call(_interp_call, func, args)
|
||||
except ExecutionFailed as exc:
|
||||
raise BrokenWorkerInterpreter(exc.excinfo) from exc
|
||||
|
||||
if is_exception:
|
||||
raise res
|
||||
|
||||
return res
|
||||
elif sys.version_info >= (3, 13):
|
||||
import _interpqueues
|
||||
import _interpreters
|
||||
|
||||
UNBOUND: Final = 2 # I have no clue how this works, but it was used in the stdlib
|
||||
FMT_UNPICKLED: Final = 0
|
||||
FMT_PICKLED: Final = 1
|
||||
QUEUE_PICKLE_ARGS: Final = (FMT_PICKLED, UNBOUND)
|
||||
QUEUE_UNPICKLE_ARGS: Final = (FMT_UNPICKLED, UNBOUND)
|
||||
|
||||
_run_func = compile(
|
||||
"""
|
||||
import _interpqueues
|
||||
from _interpreters import NotShareableError
|
||||
from pickle import loads, dumps, HIGHEST_PROTOCOL
|
||||
|
||||
QUEUE_PICKLE_ARGS = (1, 2)
|
||||
QUEUE_UNPICKLE_ARGS = (0, 2)
|
||||
|
||||
item = _interpqueues.get(queue_id)[0]
|
||||
try:
|
||||
func, args = loads(item)
|
||||
retval = func(*args)
|
||||
except BaseException as exc:
|
||||
is_exception = True
|
||||
retval = exc
|
||||
else:
|
||||
is_exception = False
|
||||
|
||||
try:
|
||||
_interpqueues.put(queue_id, (retval, is_exception), *QUEUE_UNPICKLE_ARGS)
|
||||
except NotShareableError:
|
||||
retval = dumps(retval, HIGHEST_PROTOCOL)
|
||||
_interpqueues.put(queue_id, (retval, is_exception), *QUEUE_PICKLE_ARGS)
|
||||
""",
|
||||
"<string>",
|
||||
"exec",
|
||||
)
|
||||
|
||||
class _Worker:
|
||||
last_used: float = 0
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._interpreter_id = _interpreters.create()
|
||||
self._queue_id = _interpqueues.create(1, *QUEUE_UNPICKLE_ARGS)
|
||||
_interpreters.set___main___attrs(
|
||||
self._interpreter_id, {"queue_id": self._queue_id}
|
||||
)
|
||||
|
||||
def destroy(self) -> None:
|
||||
_interpqueues.destroy(self._queue_id)
|
||||
_interpreters.destroy(self._interpreter_id)
|
||||
|
||||
def call(
|
||||
self,
|
||||
func: Callable[..., T_Retval],
|
||||
args: tuple[Any, ...],
|
||||
) -> T_Retval:
|
||||
import pickle
|
||||
|
||||
item = pickle.dumps((func, args), pickle.HIGHEST_PROTOCOL)
|
||||
_interpqueues.put(self._queue_id, item, *QUEUE_PICKLE_ARGS)
|
||||
exc_info = _interpreters.exec(self._interpreter_id, _run_func)
|
||||
if exc_info:
|
||||
raise BrokenWorkerInterpreter(exc_info)
|
||||
|
||||
res = _interpqueues.get(self._queue_id)
|
||||
(res, is_exception), fmt = res[:2]
|
||||
if fmt == FMT_PICKLED:
|
||||
res = pickle.loads(res)
|
||||
|
||||
if is_exception:
|
||||
raise res
|
||||
|
||||
return res
|
||||
else:
|
||||
|
||||
class _Worker:
|
||||
last_used: float = 0
|
||||
|
||||
def __init__(self) -> None:
|
||||
raise RuntimeError("subinterpreters require at least Python 3.13")
|
||||
|
||||
def call(
|
||||
self,
|
||||
func: Callable[..., T_Retval],
|
||||
args: tuple[Any, ...],
|
||||
) -> T_Retval:
|
||||
raise NotImplementedError
|
||||
|
||||
def destroy(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
DEFAULT_CPU_COUNT: Final = 8 # this is just an arbitrarily selected value
|
||||
MAX_WORKER_IDLE_TIME = (
|
||||
30 # seconds a subinterpreter can be idle before becoming eligible for pruning
|
||||
)
|
||||
|
||||
T_Retval = TypeVar("T_Retval")
|
||||
PosArgsT = TypeVarTuple("PosArgsT")
|
||||
|
||||
_idle_workers = RunVar[deque[_Worker]]("_available_workers")
|
||||
_default_interpreter_limiter = RunVar[CapacityLimiter]("_default_interpreter_limiter")
|
||||
|
||||
|
||||
def _stop_workers(workers: deque[_Worker]) -> None:
|
||||
for worker in workers:
|
||||
worker.destroy()
|
||||
|
||||
workers.clear()
|
||||
|
||||
|
||||
async def run_sync(
|
||||
func: Callable[[Unpack[PosArgsT]], T_Retval],
|
||||
*args: Unpack[PosArgsT],
|
||||
limiter: CapacityLimiter | None = None,
|
||||
) -> T_Retval:
|
||||
"""
|
||||
Call the given function with the given arguments in a subinterpreter.
|
||||
|
||||
.. warning:: On Python 3.13, the :mod:`concurrent.interpreters` module was not yet
|
||||
available, so the code path for that Python version relies on an undocumented,
|
||||
private API. As such, it is recommended to not rely on this function for anything
|
||||
mission-critical on Python 3.13.
|
||||
|
||||
:param func: a callable
|
||||
:param args: the positional arguments for the callable
|
||||
:param limiter: capacity limiter to use to limit the total number of subinterpreters
|
||||
running (if omitted, the default limiter is used)
|
||||
:return: the result of the call
|
||||
:raises BrokenWorkerInterpreter: if there's an internal error in a subinterpreter
|
||||
|
||||
"""
|
||||
if limiter is None:
|
||||
limiter = current_default_interpreter_limiter()
|
||||
|
||||
try:
|
||||
idle_workers = _idle_workers.get()
|
||||
except LookupError:
|
||||
idle_workers = deque()
|
||||
_idle_workers.set(idle_workers)
|
||||
atexit.register(_stop_workers, idle_workers)
|
||||
|
||||
async with limiter:
|
||||
try:
|
||||
worker = idle_workers.pop()
|
||||
except IndexError:
|
||||
worker = _Worker()
|
||||
|
||||
try:
|
||||
return await to_thread.run_sync(
|
||||
worker.call,
|
||||
func,
|
||||
args,
|
||||
limiter=limiter,
|
||||
)
|
||||
finally:
|
||||
# Prune workers that have been idle for too long
|
||||
now = current_time()
|
||||
while idle_workers:
|
||||
if now - idle_workers[0].last_used <= MAX_WORKER_IDLE_TIME:
|
||||
break
|
||||
|
||||
await to_thread.run_sync(idle_workers.popleft().destroy, limiter=limiter)
|
||||
|
||||
worker.last_used = current_time()
|
||||
idle_workers.append(worker)
|
||||
|
||||
|
||||
def current_default_interpreter_limiter() -> CapacityLimiter:
|
||||
"""
|
||||
Return the capacity limiter used by default to limit the number of concurrently
|
||||
running subinterpreters.
|
||||
|
||||
Defaults to the number of CPU cores.
|
||||
|
||||
:return: a capacity limiter object
|
||||
|
||||
"""
|
||||
try:
|
||||
return _default_interpreter_limiter.get()
|
||||
except LookupError:
|
||||
limiter = CapacityLimiter(os.cpu_count() or DEFAULT_CPU_COUNT)
|
||||
_default_interpreter_limiter.set(limiter)
|
||||
return limiter
|
||||
268
venv/Lib/site-packages/anyio/to_process.py
Normal file
268
venv/Lib/site-packages/anyio/to_process.py
Normal file
@@ -0,0 +1,268 @@
|
||||
from __future__ import annotations
|
||||
|
||||
__all__ = (
|
||||
"current_default_process_limiter",
|
||||
"process_worker",
|
||||
"run_sync",
|
||||
)
|
||||
|
||||
import os
|
||||
import pickle
|
||||
import runpy
|
||||
import subprocess
|
||||
import sys
|
||||
from collections import deque
|
||||
from collections.abc import Callable
|
||||
from types import ModuleType
|
||||
from typing import TypeVar, cast
|
||||
|
||||
from ._core._eventloop import current_time, get_async_backend, get_cancelled_exc_class
|
||||
from ._core._exceptions import BrokenWorkerProcess
|
||||
from ._core._subprocesses import open_process
|
||||
from ._core._synchronization import CapacityLimiter
|
||||
from ._core._tasks import CancelScope, fail_after
|
||||
from .abc import ByteReceiveStream, ByteSendStream, Process
|
||||
from .lowlevel import RunVar, checkpoint_if_cancelled
|
||||
from .streams.buffered import BufferedByteReceiveStream
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from typing import TypeVarTuple, Unpack
|
||||
else:
|
||||
from typing_extensions import TypeVarTuple, Unpack
|
||||
|
||||
WORKER_MAX_IDLE_TIME = 300 # 5 minutes
|
||||
|
||||
T_Retval = TypeVar("T_Retval")
|
||||
PosArgsT = TypeVarTuple("PosArgsT")
|
||||
|
||||
_process_pool_workers: RunVar[set[Process]] = RunVar("_process_pool_workers")
|
||||
_process_pool_idle_workers: RunVar[deque[tuple[Process, float]]] = RunVar(
|
||||
"_process_pool_idle_workers"
|
||||
)
|
||||
_default_process_limiter: RunVar[CapacityLimiter] = RunVar("_default_process_limiter")
|
||||
|
||||
|
||||
async def run_sync( # type: ignore[return]
|
||||
func: Callable[[Unpack[PosArgsT]], T_Retval],
|
||||
*args: Unpack[PosArgsT],
|
||||
cancellable: bool = False,
|
||||
limiter: CapacityLimiter | None = None,
|
||||
) -> T_Retval:
|
||||
"""
|
||||
Call the given function with the given arguments in a worker process.
|
||||
|
||||
If the ``cancellable`` option is enabled and the task waiting for its completion is
|
||||
cancelled, the worker process running it will be abruptly terminated using SIGKILL
|
||||
(or ``terminateProcess()`` on Windows).
|
||||
|
||||
:param func: a callable
|
||||
:param args: positional arguments for the callable
|
||||
:param cancellable: ``True`` to allow cancellation of the operation while it's
|
||||
running
|
||||
:param limiter: capacity limiter to use to limit the total amount of processes
|
||||
running (if omitted, the default limiter is used)
|
||||
:raises NoEventLoopError: if no supported asynchronous event loop is running in the
|
||||
current thread
|
||||
:return: an awaitable that yields the return value of the function.
|
||||
|
||||
"""
|
||||
|
||||
async def send_raw_command(pickled_cmd: bytes) -> object:
|
||||
try:
|
||||
await stdin.send(pickled_cmd)
|
||||
response = await buffered.receive_until(b"\n", 50)
|
||||
status, length = response.split(b" ")
|
||||
if status not in (b"RETURN", b"EXCEPTION"):
|
||||
raise RuntimeError(
|
||||
f"Worker process returned unexpected response: {response!r}"
|
||||
)
|
||||
|
||||
pickled_response = await buffered.receive_exactly(int(length))
|
||||
except BaseException as exc:
|
||||
workers.discard(process)
|
||||
try:
|
||||
process.kill()
|
||||
with CancelScope(shield=True):
|
||||
await process.aclose()
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
|
||||
if isinstance(exc, get_cancelled_exc_class()):
|
||||
raise
|
||||
else:
|
||||
raise BrokenWorkerProcess from exc
|
||||
|
||||
retval = pickle.loads(pickled_response)
|
||||
if status == b"EXCEPTION":
|
||||
assert isinstance(retval, BaseException)
|
||||
raise retval
|
||||
else:
|
||||
return retval
|
||||
|
||||
# First pickle the request before trying to reserve a worker process
|
||||
await checkpoint_if_cancelled()
|
||||
request = pickle.dumps(("run", func, args), protocol=pickle.HIGHEST_PROTOCOL)
|
||||
|
||||
# If this is the first run in this event loop thread, set up the necessary variables
|
||||
try:
|
||||
workers = _process_pool_workers.get()
|
||||
idle_workers = _process_pool_idle_workers.get()
|
||||
except LookupError:
|
||||
workers = set()
|
||||
idle_workers = deque()
|
||||
_process_pool_workers.set(workers)
|
||||
_process_pool_idle_workers.set(idle_workers)
|
||||
get_async_backend().setup_process_pool_exit_at_shutdown(workers)
|
||||
|
||||
async with limiter or current_default_process_limiter():
|
||||
# Pop processes from the pool (starting from the most recently used) until we
|
||||
# find one that hasn't exited yet
|
||||
process: Process
|
||||
while idle_workers:
|
||||
process, idle_since = idle_workers.pop()
|
||||
if process.returncode is None:
|
||||
stdin = cast(ByteSendStream, process.stdin)
|
||||
buffered = BufferedByteReceiveStream(
|
||||
cast(ByteReceiveStream, process.stdout)
|
||||
)
|
||||
|
||||
# Prune any other workers that have been idle for WORKER_MAX_IDLE_TIME
|
||||
# seconds or longer
|
||||
now = current_time()
|
||||
killed_processes: list[Process] = []
|
||||
while idle_workers:
|
||||
if now - idle_workers[0][1] < WORKER_MAX_IDLE_TIME:
|
||||
break
|
||||
|
||||
process_to_kill, idle_since = idle_workers.popleft()
|
||||
process_to_kill.kill()
|
||||
workers.remove(process_to_kill)
|
||||
killed_processes.append(process_to_kill)
|
||||
|
||||
with CancelScope(shield=True):
|
||||
for killed_process in killed_processes:
|
||||
await killed_process.aclose()
|
||||
|
||||
break
|
||||
|
||||
workers.remove(process)
|
||||
else:
|
||||
command = [sys.executable, "-u", "-m", __name__]
|
||||
process = await open_process(
|
||||
command, stdin=subprocess.PIPE, stdout=subprocess.PIPE
|
||||
)
|
||||
try:
|
||||
stdin = cast(ByteSendStream, process.stdin)
|
||||
buffered = BufferedByteReceiveStream(
|
||||
cast(ByteReceiveStream, process.stdout)
|
||||
)
|
||||
with fail_after(20):
|
||||
message = await buffered.receive(6)
|
||||
|
||||
if message != b"READY\n":
|
||||
raise BrokenWorkerProcess(
|
||||
f"Worker process returned unexpected response: {message!r}"
|
||||
)
|
||||
|
||||
main_module_path = getattr(sys.modules["__main__"], "__file__", None)
|
||||
pickled = pickle.dumps(
|
||||
("init", sys.path, main_module_path),
|
||||
protocol=pickle.HIGHEST_PROTOCOL,
|
||||
)
|
||||
await send_raw_command(pickled)
|
||||
except (BrokenWorkerProcess, get_cancelled_exc_class()):
|
||||
raise
|
||||
except BaseException as exc:
|
||||
process.kill()
|
||||
raise BrokenWorkerProcess(
|
||||
"Error during worker process initialization"
|
||||
) from exc
|
||||
|
||||
workers.add(process)
|
||||
|
||||
with CancelScope(shield=not cancellable):
|
||||
try:
|
||||
return cast(T_Retval, await send_raw_command(request))
|
||||
finally:
|
||||
if process in workers:
|
||||
idle_workers.append((process, current_time()))
|
||||
|
||||
|
||||
def current_default_process_limiter() -> CapacityLimiter:
|
||||
"""
|
||||
Return the capacity limiter that is used by default to limit the number of worker
|
||||
processes.
|
||||
|
||||
:return: a capacity limiter object
|
||||
|
||||
"""
|
||||
try:
|
||||
return _default_process_limiter.get()
|
||||
except LookupError:
|
||||
limiter = CapacityLimiter(os.cpu_count() or 2)
|
||||
_default_process_limiter.set(limiter)
|
||||
return limiter
|
||||
|
||||
|
||||
def process_worker() -> None:
|
||||
# Redirect standard streams to os.devnull so that user code won't interfere with the
|
||||
# parent-worker communication
|
||||
stdin = sys.stdin
|
||||
stdout = sys.stdout
|
||||
sys.stdin = open(os.devnull)
|
||||
sys.stdout = open(os.devnull, "w")
|
||||
|
||||
stdout.buffer.write(b"READY\n")
|
||||
while True:
|
||||
retval = exception = None
|
||||
try:
|
||||
command, *args = pickle.load(stdin.buffer)
|
||||
except EOFError:
|
||||
return
|
||||
except BaseException as exc:
|
||||
exception = exc
|
||||
else:
|
||||
if command == "run":
|
||||
func, args = args
|
||||
try:
|
||||
retval = func(*args)
|
||||
except BaseException as exc:
|
||||
exception = exc
|
||||
elif command == "init":
|
||||
main_module_path: str | None
|
||||
sys.path, main_module_path = args
|
||||
del sys.modules["__main__"]
|
||||
if main_module_path and os.path.isfile(main_module_path):
|
||||
# Load the parent's main module but as __mp_main__ instead of
|
||||
# __main__ (like multiprocessing does) to avoid infinite recursion
|
||||
try:
|
||||
main = ModuleType("__mp_main__")
|
||||
main_content = runpy.run_path(
|
||||
main_module_path, run_name="__mp_main__"
|
||||
)
|
||||
main.__dict__.update(main_content)
|
||||
sys.modules["__main__"] = sys.modules["__mp_main__"] = main
|
||||
except BaseException as exc:
|
||||
exception = exc
|
||||
try:
|
||||
if exception is not None:
|
||||
status = b"EXCEPTION"
|
||||
pickled = pickle.dumps(exception, pickle.HIGHEST_PROTOCOL)
|
||||
else:
|
||||
status = b"RETURN"
|
||||
pickled = pickle.dumps(retval, pickle.HIGHEST_PROTOCOL)
|
||||
except BaseException as exc:
|
||||
exception = exc
|
||||
status = b"EXCEPTION"
|
||||
pickled = pickle.dumps(exc, pickle.HIGHEST_PROTOCOL)
|
||||
|
||||
stdout.buffer.write(b"%s %d\n" % (status, len(pickled)))
|
||||
stdout.buffer.write(pickled)
|
||||
|
||||
# Respect SIGTERM
|
||||
if isinstance(exception, SystemExit):
|
||||
raise exception
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
process_worker()
|
||||
78
venv/Lib/site-packages/anyio/to_thread.py
Normal file
78
venv/Lib/site-packages/anyio/to_thread.py
Normal file
@@ -0,0 +1,78 @@
|
||||
from __future__ import annotations
|
||||
|
||||
__all__ = (
|
||||
"run_sync",
|
||||
"current_default_thread_limiter",
|
||||
)
|
||||
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from typing import TypeVar
|
||||
from warnings import warn
|
||||
|
||||
from ._core._eventloop import get_async_backend
|
||||
from .abc import CapacityLimiter
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from typing import TypeVarTuple, Unpack
|
||||
else:
|
||||
from typing_extensions import TypeVarTuple, Unpack
|
||||
|
||||
T_Retval = TypeVar("T_Retval")
|
||||
PosArgsT = TypeVarTuple("PosArgsT")
|
||||
|
||||
|
||||
async def run_sync(
|
||||
func: Callable[[Unpack[PosArgsT]], T_Retval],
|
||||
*args: Unpack[PosArgsT],
|
||||
abandon_on_cancel: bool = False,
|
||||
cancellable: bool | None = None,
|
||||
limiter: CapacityLimiter | None = None,
|
||||
) -> T_Retval:
|
||||
"""
|
||||
Call the given function with the given arguments in a worker thread.
|
||||
|
||||
If the ``abandon_on_cancel`` option is enabled and the task waiting for its
|
||||
completion is cancelled, the thread will still run its course but its
|
||||
return value (or any raised exception) will be ignored.
|
||||
|
||||
:param func: a callable
|
||||
:param args: positional arguments for the callable
|
||||
:param abandon_on_cancel: ``True`` to abandon the thread (leaving it to run
|
||||
unchecked on own) if the host task is cancelled, ``False`` to ignore
|
||||
cancellations in the host task until the operation has completed in the worker
|
||||
thread
|
||||
:param cancellable: deprecated alias of ``abandon_on_cancel``; will override
|
||||
``abandon_on_cancel`` if both parameters are passed
|
||||
:param limiter: capacity limiter to use to limit the total amount of threads running
|
||||
(if omitted, the default limiter is used)
|
||||
:raises NoEventLoopError: if no supported asynchronous event loop is running in the
|
||||
current thread
|
||||
:return: an awaitable that yields the return value of the function.
|
||||
|
||||
"""
|
||||
if cancellable is not None:
|
||||
abandon_on_cancel = cancellable
|
||||
warn(
|
||||
"The `cancellable=` keyword argument to `anyio.to_thread.run_sync` is "
|
||||
"deprecated since AnyIO 4.1.0; use `abandon_on_cancel=` instead",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
return await get_async_backend().run_sync_in_worker_thread(
|
||||
func, args, abandon_on_cancel=abandon_on_cancel, limiter=limiter
|
||||
)
|
||||
|
||||
|
||||
def current_default_thread_limiter() -> CapacityLimiter:
|
||||
"""
|
||||
Return the capacity limiter that is used by default to limit the number of
|
||||
concurrent threads.
|
||||
|
||||
:return: a capacity limiter object
|
||||
:raises NoEventLoopError: if no supported asynchronous event loop is running in the
|
||||
current thread
|
||||
|
||||
"""
|
||||
return get_async_backend().current_default_thread_limiter()
|
||||
Reference in New Issue
Block a user