Intermediate changes

commit_hash:f0be57c48a0eefc4e1e20805951a72528e247e44

29 files changed, 682 insertions, 599 deletions

diff --git a/contrib/libs/cxxsupp/builtins/.yandex_meta/build.ym b/contrib/libs/cxxsupp/builtins/.yandex_meta/build.ym
index 36dc94654c8..8128b2e4a3a 100644
--- a/contrib/libs/cxxsupp/builtins/.yandex_meta/build.ym
+++ b/contrib/libs/cxxsupp/builtins/.yandex_meta/build.ym
@@ -166,12 +166,6 @@ IF (ARCH_ARM64 OR ARCH_X86_64)
 ENDIF()
 EOF
 
-echo 'IF (ARCH_WASM64 OR ARCH_WASM32)'
-echo 'SRCS('
-ls wasm/*.S 
-echo ')'
-echo 'ENDIF()'
-
 echo 'IF (ARCH_ARM6 OR ARCH_ARM7)'
 echo 'SRCS('
 ls arm/*.S arm/*.c > special
diff --git a/contrib/libs/cxxsupp/builtins/ya.make b/contrib/libs/cxxsupp/builtins/ya.make
index 917316c1fc8..8f8cfe7bd86 100644
--- a/contrib/libs/cxxsupp/builtins/ya.make
+++ b/contrib/libs/cxxsupp/builtins/ya.make
@@ -95,13 +95,6 @@ IF (ARCH_ARM64 OR ARCH_X86_64)
     ENDIF()
 ENDIF()
 
-IF (ARCH_WASM64 OR ARCH_WASM32)
-    SRCS(
-        wasm/__c_longjmp.S
-        wasm/__cpp_exception.S
-    )
-ENDIF()
-
 IF (ARCH_ARM6 OR ARCH_ARM7)
     SRCS(
         absvdi2.c
diff --git a/contrib/python/aiohttp/.dist-info/METADATA b/contrib/python/aiohttp/.dist-info/METADATA
index 7ad8cef0fd3..72c03e1c7f4 100644
--- a/contrib/python/aiohttp/.dist-info/METADATA
+++ b/contrib/python/aiohttp/.dist-info/METADATA
@@ -1,6 +1,6 @@
 Metadata-Version: 2.1
 Name: aiohttp
-Version: 3.10.6
+Version: 3.10.11
 Summary: Async http client/server framework (asyncio)
 Home-page: https://github.com/aio-libs/aiohttp
 Maintainer: aiohttp team <[email protected]>
@@ -39,7 +39,7 @@ Requires-Dist: attrs >=17.3.0
 Requires-Dist: frozenlist >=1.1.1
 Requires-Dist: multidict <7.0,>=4.5
 Requires-Dist: yarl <2.0,>=1.12.0
-Requires-Dist: async-timeout <5.0,>=4.0 ; python_version < "3.11"
+Requires-Dist: async-timeout <6.0,>=4.0 ; python_version < "3.11"
 Provides-Extra: speedups
 Requires-Dist: brotlicffi ; (platform_python_implementation != "CPython") and extra == 'speedups'
 Requires-Dist: Brotli ; (platform_python_implementation == "CPython") and extra == 'speedups'
@@ -64,6 +64,10 @@ Async http client/server framework
    :target: https://codecov.io/gh/aio-libs/aiohttp
    :alt: codecov.io status for master branch
 
+.. image:: https://img.shields.io/endpoint?url=https://codspeed.io/badge.json
+   :target: https://codspeed.io/aio-libs/aiohttp
+   :alt: Codspeed.io status for aiohttp
+
 .. image:: https://badge.fury.io/py/aiohttp.svg
    :target: https://pypi.org/project/aiohttp
    :alt: Latest PyPI package version
diff --git a/contrib/python/aiohttp/README.rst b/contrib/python/aiohttp/README.rst
index 470ced9b29c..554627a42e7 100644
--- a/contrib/python/aiohttp/README.rst
+++ b/contrib/python/aiohttp/README.rst
@@ -17,6 +17,10 @@ Async http client/server framework
    :target: https://codecov.io/gh/aio-libs/aiohttp
    :alt: codecov.io status for master branch
 
+.. image:: https://img.shields.io/endpoint?url=https://codspeed.io/badge.json
+   :target: https://codspeed.io/aio-libs/aiohttp
+   :alt: Codspeed.io status for aiohttp
+
 .. image:: https://badge.fury.io/py/aiohttp.svg
    :target: https://pypi.org/project/aiohttp
    :alt: Latest PyPI package version
diff --git a/contrib/python/aiohttp/aiohttp/__init__.py b/contrib/python/aiohttp/aiohttp/__init__.py
index 8830d340940..4fc7908843a 100644
--- a/contrib/python/aiohttp/aiohttp/__init__.py
+++ b/contrib/python/aiohttp/aiohttp/__init__.py
@@ -1,4 +1,4 @@
-__version__ = "3.10.6"
+__version__ = "3.10.11"
 
 from typing import TYPE_CHECKING, Tuple
 
@@ -8,6 +8,7 @@ from .client import (
     ClientConnectionError,
     ClientConnectionResetError,
     ClientConnectorCertificateError,
+    ClientConnectorDNSError,
     ClientConnectorError,
     ClientConnectorSSLError,
     ClientError,
@@ -127,6 +128,7 @@ __all__: Tuple[str, ...] = (
     "ClientConnectionError",
     "ClientConnectionResetError",
     "ClientConnectorCertificateError",
+    "ClientConnectorDNSError",
     "ClientConnectorError",
     "ClientConnectorSSLError",
     "ClientError",
diff --git a/contrib/python/aiohttp/aiohttp/_http_writer.pyx b/contrib/python/aiohttp/aiohttp/_http_writer.pyx
index eff85219586..d19c20d76cc 100644
--- a/contrib/python/aiohttp/aiohttp/_http_writer.pyx
+++ b/contrib/python/aiohttp/aiohttp/_http_writer.pyx
@@ -127,10 +127,6 @@ def _serialize_headers(str status_line, headers):
 
     _init_writer(&writer)
 
-    for key, val in headers.items():
-        _safe_header(to_str(key))
-        _safe_header(to_str(val))
-
     try:
         if _write_str(&writer, status_line) < 0:
             raise
@@ -140,6 +136,9 @@ def _serialize_headers(str status_line, headers):
             raise
 
         for key, val in headers.items():
+            _safe_header(to_str(key))
+            _safe_header(to_str(val))
+
             if _write_str(&writer, to_str(key)) < 0:
                 raise
             if _write_byte(&writer, b':') < 0:
diff --git a/contrib/python/aiohttp/aiohttp/client.py b/contrib/python/aiohttp/aiohttp/client.py
index 19d27eca31a..9c7ca13fb7d 100644
--- a/contrib/python/aiohttp/aiohttp/client.py
+++ b/contrib/python/aiohttp/aiohttp/client.py
@@ -42,6 +42,7 @@ from .client_exceptions import (
     ClientConnectionError,
     ClientConnectionResetError,
     ClientConnectorCertificateError,
+    ClientConnectorDNSError,
     ClientConnectorError,
     ClientConnectorSSLError,
     ClientError,
@@ -88,7 +89,6 @@ from .helpers import (
     DEBUG,
     BasicAuth,
     TimeoutHandle,
-    ceil_timeout,
     get_env_proxy_for_url,
     method_must_be_empty_body,
     sentinel,
@@ -105,6 +105,7 @@ __all__ = (
     "ClientConnectionError",
     "ClientConnectionResetError",
     "ClientConnectorCertificateError",
+    "ClientConnectorDNSError",
     "ClientConnectorError",
     "ClientConnectorSSLError",
     "ClientError",
@@ -208,7 +209,7 @@ class ClientTimeout:
 
 
 # 5 Minute default read timeout
-DEFAULT_TIMEOUT: Final[ClientTimeout] = ClientTimeout(total=5 * 60)
+DEFAULT_TIMEOUT: Final[ClientTimeout] = ClientTimeout(total=5 * 60, sock_connect=30)
 
 # https://www.rfc-editor.org/rfc/rfc9110#section-9.2.2
 IDEMPOTENT_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE", "PUT", "DELETE"})
@@ -354,7 +355,7 @@ class ClientSession:
             cookie_jar = CookieJar(loop=loop)
         self._cookie_jar = cookie_jar
 
-        if cookies is not None:
+        if cookies:
             self._cookie_jar.update_cookies(cookies)
 
         self._connector = connector
@@ -444,7 +445,7 @@ class ClientSession:
         if self._base_url is None:
             return url
         else:
-            assert not url.is_absolute() and url.path.startswith("/")
+            assert not url.absolute and url.path.startswith("/")
             return self._base_url.join(url)
 
     async def _request(
@@ -477,7 +478,7 @@ class ClientSession:
         ssl: Union[SSLContext, bool, Fingerprint] = True,
         server_hostname: Optional[str] = None,
         proxy_headers: Optional[LooseHeaders] = None,
-        trace_request_ctx: Optional[Mapping[str, str]] = None,
+        trace_request_ctx: Optional[Mapping[str, Any]] = None,
         read_bufsize: Optional[int] = None,
         auto_decompress: Optional[bool] = None,
         max_line_size: Optional[int] = None,
@@ -504,13 +505,12 @@ class ClientSession:
             warnings.warn("Chunk size is deprecated #1615", DeprecationWarning)
 
         redirects = 0
-        history = []
+        history: List[ClientResponse] = []
         version = self._version
         params = params or {}
 
         # Merge with default headers and transform to CIMultiDict
         headers = self._prepare_headers(headers)
-        proxy_headers = self._prepare_headers(proxy_headers)
 
         try:
             url = self._build_url(str_or_url)
@@ -526,7 +526,10 @@ class ClientSession:
             for i in skip_auto_headers:
                 skip_headers.add(istr(i))
 
-        if proxy is not None:
+        if proxy is None:
+            proxy_headers = None
+        else:
+            proxy_headers = self._prepare_headers(proxy_headers)
             try:
                 proxy = URL(proxy)
             except ValueError as e:
@@ -586,13 +589,18 @@ class ClientSession:
                             else InvalidUrlClientError
                         )
                         raise err_exc_cls(url)
-                    if auth and auth_from_url:
+                    # If `auth` was passed for an already authenticated URL,
+                    # disallow only if this is the initial URL; this is to avoid issues
+                    # with sketchy redirects that are not the caller's responsibility
+                    if not history and (auth and auth_from_url):
                         raise ValueError(
                             "Cannot combine AUTH argument with "
                             "credentials encoded in URL"
                         )
 
-                    if auth is None:
+                    # Override the auth with the one from the URL only if we
+                    # have no auth, or if we got an auth from a redirect URL
+                    if auth is None or (history and auth_from_url is not None):
                         auth = auth_from_url
                     if auth is None:
                         auth = self._default_auth
@@ -652,13 +660,9 @@ class ClientSession:
 
                     # connection timeout
                     try:
-                        async with ceil_timeout(
-                            real_timeout.connect,
-                            ceil_threshold=real_timeout.ceil_threshold,
-                        ):
-                            conn = await self._connector.connect(
-                                req, traces=traces, timeout=real_timeout
-                            )
+                        conn = await self._connector.connect(
+                            req, traces=traces, timeout=real_timeout
+                        )
                     except asyncio.TimeoutError as exc:
                         raise ConnectionTimeoutError(
                             f"Connection timeout to host {url}"
@@ -702,7 +706,8 @@ class ClientSession:
                             raise
                         raise ClientOSError(*exc.args) from exc
 
-                    self._cookie_jar.update_cookies(resp.cookies, resp.url)
+                    if cookies := resp.cookies:
+                        self._cookie_jar.update_cookies(cookies, resp.url)
 
                     # redirects
                     if resp.status in (301, 302, 303, 307, 308) and allow_redirects:
diff --git a/contrib/python/aiohttp/aiohttp/client_exceptions.py b/contrib/python/aiohttp/aiohttp/client_exceptions.py
index 94991c42477..2cf6cf88328 100644
--- a/contrib/python/aiohttp/aiohttp/client_exceptions.py
+++ b/contrib/python/aiohttp/aiohttp/client_exceptions.py
@@ -30,6 +30,7 @@ __all__ = (
     "ClientConnectorError",
     "ClientProxyConnectionError",
     "ClientSSLError",
+    "ClientConnectorDNSError",
     "ClientConnectorSSLError",
     "ClientConnectorCertificateError",
     "ConnectionTimeoutError",
@@ -206,6 +207,14 @@ class ClientConnectorError(ClientOSError):
     __reduce__ = BaseException.__reduce__
 
 
+class ClientConnectorDNSError(ClientConnectorError):
+    """DNS resolution failed during client connection.
+
+    Raised in :class:`aiohttp.connector.TCPConnector` if
+        DNS resolution fails.
+    """
+
+
 class ClientProxyConnectionError(ClientConnectorError):
     """Proxy connection error.
 
diff --git a/contrib/python/aiohttp/aiohttp/client_reqrep.py b/contrib/python/aiohttp/aiohttp/client_reqrep.py
index aa8f54e67b8..91605f0e83d 100644
--- a/contrib/python/aiohttp/aiohttp/client_reqrep.py
+++ b/contrib/python/aiohttp/aiohttp/client_reqrep.py
@@ -3,6 +3,7 @@ import codecs
 import contextlib
 import functools
 import io
+import itertools
 import re
 import sys
 import traceback
@@ -89,8 +90,11 @@ if TYPE_CHECKING:
 
 
 _CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
-_YARL_SUPPORTS_EXTEND_QUERY = tuple(map(int, yarl_version.split(".")[:2])) >= (1, 11)
 json_re = re.compile(r"^application/(?:[\w.+-]+?\+)?json")
+_YARL_SUPPORTS_HOST_SUBCOMPONENT = tuple(map(int, yarl_version.split(".")[:2])) >= (
+    1,
+    13,
+)
 
 
 def _gen_default_accept_encoding() -> str:
@@ -241,6 +245,9 @@ class ClientRequest:
     }
     POST_METHODS = {hdrs.METH_PATCH, hdrs.METH_POST, hdrs.METH_PUT}
     ALL_METHODS = GET_METHODS.union(POST_METHODS).union({hdrs.METH_DELETE})
+    _HOST_STRINGS = frozenset(
+        map("".join, itertools.product(*zip("host".upper(), "host".lower())))
+    )
 
     DEFAULT_HEADERS = {
         hdrs.ACCEPT: "*/*",
@@ -289,29 +296,24 @@ class ClientRequest:
     ):
         if loop is None:
             loop = asyncio.get_event_loop()
-
-        match = _CONTAINS_CONTROL_CHAR_RE.search(method)
-        if match:
+        if match := _CONTAINS_CONTROL_CHAR_RE.search(method):
             raise ValueError(
                 f"Method cannot contain non-token characters {method!r} "
-                "(found at least {match.group()!r})"
+                f"(found at least {match.group()!r})"
             )
-
-        assert isinstance(url, URL), url
-        assert isinstance(proxy, (URL, type(None))), proxy
+        # URL forbids subclasses, so a simple type check is enough.
+        assert type(url) is URL, url
+        if proxy is not None:
+            assert type(proxy) is URL, proxy
         # FIXME: session is None in tests only, need to fix tests
         # assert session is not None
-        self._session = cast("ClientSession", session)
+        if TYPE_CHECKING:
+            assert session is not None
+        self._session = session
         if params:
-            if _YARL_SUPPORTS_EXTEND_QUERY:
-                url = url.extend_query(params)
-            else:
-                q = MultiDict(url.query)
-                url2 = url.with_query(params)
-                q.extend(url2.query)
-                url = url.with_query(q)
+            url = url.extend_query(params)
         self.original_url = url
-        self.url = url.with_fragment(None)
+        self.url = url.with_fragment(None) if url.raw_fragment else url
         self.method = method.upper()
         self.chunked = chunked
         self.compress = compress
@@ -342,9 +344,7 @@ class ClientRequest:
         if data is not None or self.method not in self.GET_METHODS:
             self.update_transfer_encoding()
         self.update_expect_continue(expect100)
-        if traces is None:
-            traces = []
-        self._traces = traces
+        self._traces = [] if traces is None else traces
 
     def __reset_writer(self, _: object = None) -> None:
         self.__writer = None
@@ -354,17 +354,11 @@ class ClientRequest:
         return self.__writer
 
     @_writer.setter
-    def _writer(self, writer: Optional["asyncio.Task[None]"]) -> None:
+    def _writer(self, writer: "asyncio.Task[None]") -> None:
         if self.__writer is not None:
             self.__writer.remove_done_callback(self.__reset_writer)
         self.__writer = writer
-        if writer is None:
-            return
-        if writer.done():
-            # The writer is already done, so we can reset it immediately.
-            self.__reset_writer()
-        else:
-            writer.add_done_callback(self.__reset_writer)
+        writer.add_done_callback(self.__reset_writer)
 
     def is_ssl(self) -> bool:
         return self.url.scheme in ("https", "wss")
@@ -377,7 +371,7 @@ class ClientRequest:
     def connection_key(self) -> ConnectionKey:
         proxy_headers = self.proxy_headers
         if proxy_headers:
-            h: Optional[int] = hash(tuple((k, v) for k, v in proxy_headers.items()))
+            h: Optional[int] = hash(tuple(proxy_headers.items()))
         else:
             h = None
         return ConnectionKey(
@@ -412,9 +406,8 @@ class ClientRequest:
             raise InvalidURL(url)
 
         # basic auth info
-        username, password = url.user, url.password
-        if username or password:
-            self.auth = helpers.BasicAuth(username or "", password or "")
+        if url.raw_user or url.raw_password:
+            self.auth = helpers.BasicAuth(url.user or "", url.password or "")
 
     def update_version(self, version: Union[http.HttpVersion, str]) -> None:
         """Convert request version to two elements tuple.
@@ -435,26 +428,49 @@ class ClientRequest:
         """Update request headers."""
         self.headers: CIMultiDict[str] = CIMultiDict()
 
-        # add host
-        netloc = cast(str, self.url.raw_host)
-        if helpers.is_ipv6_address(netloc):
-            netloc = f"[{netloc}]"
-        # See https://github.com/aio-libs/aiohttp/issues/3636.
-        netloc = netloc.rstrip(".")
-        if self.url.port is not None and not self.url.is_default_port():
-            netloc += ":" + str(self.url.port)
-        self.headers[hdrs.HOST] = netloc
+        # Build the host header
+        if _YARL_SUPPORTS_HOST_SUBCOMPONENT:
+            host = self.url.host_subcomponent
+            # host_subcomponent is None when the URL is a relative URL.
+            # but we know we do not have a relative URL here.
+            assert host is not None
+        else:
+            host = cast(str, self.url.raw_host)
+            if helpers.is_ipv6_address(host):
+                host = f"[{host}]"
 
-        if headers:
-            if isinstance(headers, (dict, MultiDictProxy, MultiDict)):
-                headers = headers.items()
+        if host[-1] == ".":
+            # Remove all trailing dots from the netloc as while
+            # they are valid FQDNs in DNS, TLS validation fails.
+            # See https://github.com/aio-libs/aiohttp/issues/3636.
+            # To avoid string manipulation we only call rstrip if
+            # the last character is a dot.
+            host = host.rstrip(".")
 
-            for key, value in headers:  # type: ignore[misc]
-                # A special case for Host header
-                if key.lower() == "host":
-                    self.headers[key] = value
-                else:
-                    self.headers.add(key, value)
+        # If explicit port is not None, it means that the port was
+        # explicitly specified in the URL. In this case we check
+        # if its not the default port for the scheme and add it to
+        # the host header. We check explicit_port first because
+        # yarl caches explicit_port and its likely to already be
+        # in the cache and non-default port URLs are far less common.
+        explicit_port = self.url.explicit_port
+        if explicit_port is not None and not self.url.is_default_port():
+            host = f"{host}:{explicit_port}"
+
+        self.headers[hdrs.HOST] = host
+
+        if not headers:
+            return
+
+        if isinstance(headers, (dict, MultiDictProxy, MultiDict)):
+            headers = headers.items()
+
+        for key, value in headers:  # type: ignore[misc]
+            # A special case for Host header
+            if key in self._HOST_STRINGS:
+                self.headers[key] = value
+            else:
+                self.headers.add(key, value)
 
     def update_auto_headers(self, skip_auto_headers: Optional[Iterable[str]]) -> None:
         if skip_auto_headers is not None:
@@ -592,7 +608,10 @@ class ClientRequest:
     def update_expect_continue(self, expect: bool = False) -> None:
         if expect:
             self.headers[hdrs.EXPECT] = "100-continue"
-        elif self.headers.get(hdrs.EXPECT, "").lower() == "100-continue":
+        elif (
+            hdrs.EXPECT in self.headers
+            and self.headers[hdrs.EXPECT].lower() == "100-continue"
+        ):
             expect = True
 
         if expect:
@@ -604,10 +623,16 @@ class ClientRequest:
         proxy_auth: Optional[BasicAuth],
         proxy_headers: Optional[LooseHeaders],
     ) -> None:
+        self.proxy = proxy
+        if proxy is None:
+            self.proxy_auth = None
+            self.proxy_headers = None
+            return
+
         if proxy_auth and not isinstance(proxy_auth, helpers.BasicAuth):
             raise ValueError("proxy_auth must be None or BasicAuth() tuple")
-        self.proxy = proxy
         self.proxy_auth = proxy_auth
+
         if proxy_headers is not None and not isinstance(
             proxy_headers, (MultiDict, MultiDictProxy)
         ):
@@ -619,10 +644,8 @@ class ClientRequest:
             # keep alive not supported at all
             return False
         if self.version == HttpVersion10:
-            if self.headers.get(hdrs.CONNECTION) == "keep-alive":
-                return True
-            else:  # no headers means we close for Http 1.0
-                return False
+            # no headers means we close for Http 1.0
+            return self.headers.get(hdrs.CONNECTION) == "keep-alive"
         elif self.headers.get(hdrs.CONNECTION) == "close":
             return False
 
@@ -683,17 +706,19 @@ class ClientRequest:
         # - not CONNECT proxy must send absolute form URI
         # - most common is origin form URI
         if self.method == hdrs.METH_CONNECT:
-            connect_host = self.url.raw_host
-            assert connect_host is not None
-            if helpers.is_ipv6_address(connect_host):
-                connect_host = f"[{connect_host}]"
+            if _YARL_SUPPORTS_HOST_SUBCOMPONENT:
+                connect_host = self.url.host_subcomponent
+                assert connect_host is not None
+            else:
+                connect_host = self.url.raw_host
+                assert connect_host is not None
+                if helpers.is_ipv6_address(connect_host):
+                    connect_host = f"[{connect_host}]"
             path = f"{connect_host}:{self.url.port}"
         elif self.proxy and not self.is_ssl():
             path = str(self.url)
         else:
-            path = self.url.raw_path
-            if self.url.raw_query_string:
-                path += "?" + self.url.raw_query_string
+            path = self.url.raw_path_qs
 
         protocol = conn.protocol
         assert protocol is not None
@@ -745,6 +770,7 @@ class ClientRequest:
         await writer.write_headers(status_line, self.headers)
         coro = self.write_bytes(writer, conn)
 
+        task: Optional["asyncio.Task[None]"]
         if sys.version_info >= (3, 12):
             # Optimization for Python 3.12, try to write
             # bytes immediately to avoid having to schedule
@@ -753,13 +779,17 @@ class ClientRequest:
         else:
             task = self.loop.create_task(coro)
 
-        self._writer = task
+        if task.done():
+            task = None
+        else:
+            self._writer = task
+
         response_class = self.response_class
         assert response_class is not None
         self.response = response_class(
             self.method,
             self.original_url,
-            writer=self._writer,
+            writer=task,
             continue100=self._continue,
             timer=self._timer,
             request_info=self.request_info,
@@ -770,9 +800,9 @@ class ClientRequest:
         return self.response
 
     async def close(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             try:
-                await self._writer
+                await self.__writer
             except asyncio.CancelledError:
                 if (
                     sys.version_info >= (3, 11)
@@ -782,11 +812,11 @@ class ClientRequest:
                     raise
 
     def terminate(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             if not self.loop.is_closed():
-                self._writer.cancel()
-            self._writer.remove_done_callback(self.__reset_writer)
-            self._writer = None
+                self.__writer.cancel()
+            self.__writer.remove_done_callback(self.__reset_writer)
+            self.__writer = None
 
     async def _on_chunk_request_sent(self, method: str, url: URL, chunk: bytes) -> None:
         for trace in self._traces:
@@ -799,6 +829,9 @@ class ClientRequest:
             await trace.send_request_headers(method, url, headers)
 
 
+_CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
+
+
 class ClientResponse(HeadersMixin):
 
     # Some of these attributes are None when created,
@@ -827,7 +860,7 @@ class ClientResponse(HeadersMixin):
         method: str,
         url: URL,
         *,
-        writer: "asyncio.Task[None]",
+        writer: "Optional[asyncio.Task[None]]",
         continue100: Optional["asyncio.Future[bool]"],
         timer: BaseTimerContext,
         request_info: RequestInfo,
@@ -841,9 +874,10 @@ class ClientResponse(HeadersMixin):
         self.cookies = SimpleCookie()
 
         self._real_url = url
-        self._url = url.with_fragment(None)
-        self._body: Any = None
-        self._writer: Optional[asyncio.Task[None]] = writer
+        self._url = url.with_fragment(None) if url.raw_fragment else url
+        self._body: Optional[bytes] = None
+        if writer is not None:
+            self._writer = writer
         self._continue = continue100  # None by default
         self._closed = True
         self._history: Tuple[ClientResponse, ...] = ()
@@ -871,18 +905,24 @@ class ClientResponse(HeadersMixin):
 
     @property
     def _writer(self) -> Optional["asyncio.Task[None]"]:
+        """The writer task for streaming data.
+
+        _writer is only provided for backwards compatibility
+        for subclasses that may need to access it.
+        """
         return self.__writer
 
     @_writer.setter
     def _writer(self, writer: Optional["asyncio.Task[None]"]) -> None:
+        """Set the writer task for streaming data."""
         if self.__writer is not None:
             self.__writer.remove_done_callback(self.__reset_writer)
         self.__writer = writer
         if writer is None:
             return
         if writer.done():
-            # The writer is already done, so we can reset it immediately.
-            self.__reset_writer()
+            # The writer is already done, so we can clear it immediately.
+            self.__writer = None
         else:
             writer.add_done_callback(self.__reset_writer)
 
@@ -1125,16 +1165,16 @@ class ClientResponse(HeadersMixin):
 
     def _release_connection(self) -> None:
         if self._connection is not None:
-            if self._writer is None:
+            if self.__writer is None:
                 self._connection.release()
                 self._connection = None
             else:
-                self._writer.add_done_callback(lambda f: self._release_connection())
+                self.__writer.add_done_callback(lambda f: self._release_connection())
 
     async def _wait_released(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             try:
-                await self._writer
+                await self.__writer
             except asyncio.CancelledError:
                 if (
                     sys.version_info >= (3, 11)
@@ -1145,20 +1185,20 @@ class ClientResponse(HeadersMixin):
         self._release_connection()
 
     def _cleanup_writer(self) -> None:
-        if self._writer is not None:
-            self._writer.cancel()
+        if self.__writer is not None:
+            self.__writer.cancel()
         self._session = None
 
     def _notify_content(self) -> None:
         content = self.content
         if content and content.exception() is None:
-            set_exception(content, ClientConnectionError("Connection closed"))
+            set_exception(content, _CONNECTION_CLOSED_EXCEPTION)
         self._released = True
 
     async def wait_for_close(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             try:
-                await self._writer
+                await self.__writer
             except asyncio.CancelledError:
                 if (
                     sys.version_info >= (3, 11)
@@ -1186,7 +1226,7 @@ class ClientResponse(HeadersMixin):
         protocol = self._connection and self._connection.protocol
         if protocol is None or not protocol.upgraded:
             await self._wait_released()  # Underlying connection released
-        return self._body  # type: ignore[no-any-return]
+        return self._body
 
     def get_encoding(self) -> str:
         ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
@@ -1219,9 +1259,7 @@ class ClientResponse(HeadersMixin):
         if encoding is None:
             encoding = self.get_encoding()
 
-        return self._body.decode(  # type: ignore[no-any-return,union-attr]
-            encoding, errors=errors
-        )
+        return self._body.decode(encoding, errors=errors)  # type: ignore[union-attr]
 
     async def json(
         self,
diff --git a/contrib/python/aiohttp/aiohttp/client_ws.py b/contrib/python/aiohttp/aiohttp/client_ws.py
index c6b5da5103b..96e2c59715c 100644
--- a/contrib/python/aiohttp/aiohttp/client_ws.py
+++ b/contrib/python/aiohttp/aiohttp/client_ws.py
@@ -249,10 +249,30 @@ class ClientWebSocketResponse:
             self._reader.feed_data(WS_CLOSING_MESSAGE, 0)
             await self._close_wait
 
-        if not self._closed:
-            self._set_closed()
+        if self._closed:
+            return False
+
+        self._set_closed()
+        try:
+            await self._writer.close(code, message)
+        except asyncio.CancelledError:
+            self._close_code = WSCloseCode.ABNORMAL_CLOSURE
+            self._response.close()
+            raise
+        except Exception as exc:
+            self._close_code = WSCloseCode.ABNORMAL_CLOSURE
+            self._exception = exc
+            self._response.close()
+            return True
+
+        if self._close_code:
+            self._response.close()
+            return True
+
+        while True:
             try:
-                await self._writer.close(code, message)
+                async with async_timeout.timeout(self._timeout):
+                    msg = await self._reader.read()
             except asyncio.CancelledError:
                 self._close_code = WSCloseCode.ABNORMAL_CLOSURE
                 self._response.close()
@@ -263,31 +283,11 @@ class ClientWebSocketResponse:
                 self._response.close()
                 return True
 
-            if self._close_code:
+            if msg.type is WSMsgType.CLOSE:
+                self._close_code = msg.data
                 self._response.close()
                 return True
 
-            while True:
-                try:
-                    async with async_timeout.timeout(self._timeout):
-                        msg = await self._reader.read()
-                except asyncio.CancelledError:
-                    self._close_code = WSCloseCode.ABNORMAL_CLOSURE
-                    self._response.close()
-                    raise
-                except Exception as exc:
-                    self._close_code = WSCloseCode.ABNORMAL_CLOSURE
-                    self._exception = exc
-                    self._response.close()
-                    return True
-
-                if msg.type is WSMsgType.CLOSE:
-                    self._close_code = msg.data
-                    self._response.close()
-                    return True
-        else:
-            return False
-
     async def receive(self, timeout: Optional[float] = None) -> WSMessage:
         receive_timeout = timeout or self._receive_timeout
 
diff --git a/contrib/python/aiohttp/aiohttp/compression_utils.py b/contrib/python/aiohttp/aiohttp/compression_utils.py
index ab4a2f1cc84..ebe8857f487 100644
--- a/contrib/python/aiohttp/aiohttp/compression_utils.py
+++ b/contrib/python/aiohttp/aiohttp/compression_utils.py
@@ -70,6 +70,14 @@ class ZLibCompressor(ZlibBaseHandler):
         return self._compressor.compress(data)
 
     async def compress(self, data: bytes) -> bytes:
+        """Compress the data and returned the compressed bytes.
+
+        Note that flush() must be called after the last call to compress()
+
+        If the data size is large than the max_sync_chunk_size, the compression
+        will be done in the executor. Otherwise, the compression will be done
+        in the event loop.
+        """
         async with self._compress_lock:
             # To ensure the stream is consistent in the event
             # there are multiple writers, we need to lock
@@ -79,8 +87,8 @@ class ZLibCompressor(ZlibBaseHandler):
                 self._max_sync_chunk_size is not None
                 and len(data) > self._max_sync_chunk_size
             ):
-                return await asyncio.get_event_loop().run_in_executor(
-                    self._executor, self.compress_sync, data
+                return await asyncio.get_running_loop().run_in_executor(
+                    self._executor, self._compressor.compress, data
                 )
             return self.compress_sync(data)
 
@@ -107,12 +115,18 @@ class ZLibDecompressor(ZlibBaseHandler):
         return self._decompressor.decompress(data, max_length)
 
     async def decompress(self, data: bytes, max_length: int = 0) -> bytes:
+        """Decompress the data and return the decompressed bytes.
+
+        If the data size is large than the max_sync_chunk_size, the decompression
+        will be done in the executor. Otherwise, the decompression will be done
+        in the event loop.
+        """
         if (
             self._max_sync_chunk_size is not None
             and len(data) > self._max_sync_chunk_size
         ):
-            return await asyncio.get_event_loop().run_in_executor(
-                self._executor, self.decompress_sync, data, max_length
+            return await asyncio.get_running_loop().run_in_executor(
+                self._executor, self._decompressor.decompress, data, max_length
             )
         return self.decompress_sync(data, max_length)
 
diff --git a/contrib/python/aiohttp/aiohttp/connector.py b/contrib/python/aiohttp/aiohttp/connector.py
index 8288a0115b7..7d0160fa038 100644
--- a/contrib/python/aiohttp/aiohttp/connector.py
+++ b/contrib/python/aiohttp/aiohttp/connector.py
@@ -5,11 +5,10 @@ import socket
 import sys
 import traceback
 import warnings
-from collections import defaultdict, deque
+from collections import OrderedDict, defaultdict
 from contextlib import suppress
 from http import HTTPStatus
-from http.cookies import SimpleCookie
-from itertools import cycle, islice
+from itertools import chain, cycle, islice
 from time import monotonic
 from types import TracebackType
 from typing import (
@@ -39,6 +38,7 @@ from .abc import AbstractResolver, ResolveResult
 from .client_exceptions import (
     ClientConnectionError,
     ClientConnectorCertificateError,
+    ClientConnectorDNSError,
     ClientConnectorError,
     ClientConnectorSSLError,
     ClientHttpProxyError,
@@ -50,8 +50,14 @@ from .client_exceptions import (
 )
 from .client_proto import ResponseHandler
 from .client_reqrep import ClientRequest, Fingerprint, _merge_ssl_params
-from .helpers import ceil_timeout, is_ip_address, noop, sentinel
-from .locks import EventResultOrError
+from .helpers import (
+    ceil_timeout,
+    is_ip_address,
+    noop,
+    sentinel,
+    set_exception,
+    set_result,
+)
 from .resolver import DefaultResolver
 
 try:
@@ -70,6 +76,14 @@ WS_SCHEMA_SET = frozenset({"ws", "wss"})
 HTTP_AND_EMPTY_SCHEMA_SET = HTTP_SCHEMA_SET | EMPTY_SCHEMA_SET
 HIGH_LEVEL_SCHEMA_SET = HTTP_AND_EMPTY_SCHEMA_SET | WS_SCHEMA_SET
 
+NEEDS_CLEANUP_CLOSED = (3, 13, 0) <= sys.version_info < (
+    3,
+    13,
+    1,
+) or sys.version_info < (3, 12, 7)
+# Cleanup closed is no longer needed after https://github.com/python/cpython/pull/118960
+# which first appeared in Python 3.12.7 and 3.13.1
+
 
 __all__ = ("BaseConnector", "TCPConnector", "UnixConnector", "NamedPipeConnector")
 
@@ -103,7 +117,6 @@ class _DeprecationWaiter:
 class Connection:
 
     _source_traceback = None
-    _transport = None
 
     def __init__(
         self,
@@ -181,9 +194,7 @@ class Connection:
         self._notify_release()
 
         if self._protocol is not None:
-            self._connector._release(
-                self._key, self._protocol, should_close=self._protocol.should_close
-            )
+            self._connector._release(self._key, self._protocol)
             self._protocol = None
 
     @property
@@ -194,8 +205,10 @@ class Connection:
 class _TransportPlaceholder:
     """placeholder for BaseConnector.connect function"""
 
+    __slots__ = ()
+
     def close(self) -> None:
-        pass
+        """Close the placeholder transport."""
 
 
 class BaseConnector:
@@ -260,18 +273,31 @@ class BaseConnector:
         self._force_close = force_close
 
         # {host_key: FIFO list of waiters}
-        self._waiters = defaultdict(deque)  # type: ignore[var-annotated]
+        # The FIFO is implemented with an OrderedDict with None keys because
+        # python does not have an ordered set.
+        self._waiters: DefaultDict[
+            ConnectionKey, OrderedDict[asyncio.Future[None], None]
+        ] = defaultdict(OrderedDict)
 
         self._loop = loop
         self._factory = functools.partial(ResponseHandler, loop=loop)
 
-        self.cookies = SimpleCookie()
-
         # start keep-alive connection cleanup task
         self._cleanup_handle: Optional[asyncio.TimerHandle] = None
 
         # start cleanup closed transports task
         self._cleanup_closed_handle: Optional[asyncio.TimerHandle] = None
+
+        if enable_cleanup_closed and not NEEDS_CLEANUP_CLOSED:
+            warnings.warn(
+                "enable_cleanup_closed ignored because "
+                "https://github.com/python/cpython/pull/118960 is fixed "
+                f"in Python version {sys.version_info}",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            enable_cleanup_closed = False
+
         self._cleanup_closed_disabled = not enable_cleanup_closed
         self._cleanup_closed_transports: List[Optional[asyncio.Transport]] = []
         self._cleanup_closed()
@@ -350,28 +376,22 @@ class BaseConnector:
             # recreate it ever!
             self._cleanup_handle = None
 
-        now = self._loop.time()
+        now = monotonic()
         timeout = self._keepalive_timeout
 
         if self._conns:
             connections = {}
             deadline = now - timeout
             for key, conns in self._conns.items():
-                alive = []
+                alive: List[Tuple[ResponseHandler, float]] = []
                 for proto, use_time in conns:
-                    if proto.is_connected():
-                        if use_time - deadline < 0:
-                            transport = proto.transport
-                            proto.close()
-                            if key.is_ssl and not self._cleanup_closed_disabled:
-                                self._cleanup_closed_transports.append(transport)
-                        else:
-                            alive.append((proto, use_time))
-                    else:
-                        transport = proto.transport
-                        proto.close()
-                        if key.is_ssl and not self._cleanup_closed_disabled:
-                            self._cleanup_closed_transports.append(transport)
+                    if proto.is_connected() and use_time - deadline >= 0:
+                        alive.append((proto, use_time))
+                        continue
+                    transport = proto.transport
+                    proto.close()
+                    if not self._cleanup_closed_disabled and key.is_ssl:
+                        self._cleanup_closed_transports.append(transport)
 
                 if alive:
                     connections[key] = alive
@@ -387,17 +407,6 @@ class BaseConnector:
                 timeout_ceil_threshold=self._timeout_ceil_threshold,
             )
 
-    def _drop_acquired_per_host(
-        self, key: "ConnectionKey", val: ResponseHandler
-    ) -> None:
-        acquired_per_host = self._acquired_per_host
-        if key not in acquired_per_host:
-            return
-        conns = acquired_per_host[key]
-        conns.remove(val)
-        if not conns:
-            del self._acquired_per_host[key]
-
     def _cleanup_closed(self) -> None:
         """Double confirmation for transport close.
 
@@ -458,6 +467,9 @@ class BaseConnector:
         finally:
             self._conns.clear()
             self._acquired.clear()
+            for keyed_waiters in self._waiters.values():
+                for keyed_waiter in keyed_waiters:
+                    keyed_waiter.cancel()
             self._waiters.clear()
             self._cleanup_handle = None
             self._cleanup_closed_transports.clear()
@@ -480,139 +492,153 @@ class BaseConnector:
         If it returns less than 1 means that there are no connections
         available.
         """
-        if self._limit:
-            # total calc available connections
-            available = self._limit - len(self._acquired)
+        # check total available connections
+        # If there are no limits, this will always return 1
+        total_remain = 1
 
-            # check limit per host
-            if (
-                self._limit_per_host
-                and available > 0
-                and key in self._acquired_per_host
-            ):
-                acquired = self._acquired_per_host.get(key)
-                assert acquired is not None
-                available = self._limit_per_host - len(acquired)
+        if self._limit and (total_remain := self._limit - len(self._acquired)) <= 0:
+            return total_remain
 
-        elif self._limit_per_host and key in self._acquired_per_host:
-            # check limit per host
-            acquired = self._acquired_per_host.get(key)
-            assert acquired is not None
-            available = self._limit_per_host - len(acquired)
-        else:
-            available = 1
+        # check limit per host
+        if host_remain := self._limit_per_host:
+            if acquired := self._acquired_per_host.get(key):
+                host_remain -= len(acquired)
+            if total_remain > host_remain:
+                return host_remain
 
-        return available
+        return total_remain
 
     async def connect(
         self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
     ) -> Connection:
         """Get from pool or create new connection."""
         key = req.connection_key
-        available = self._available_connections(key)
-
-        # Wait if there are no available connections or if there are/were
-        # waiters (i.e. don't steal connection from a waiter about to wake up)
-        if available <= 0 or key in self._waiters:
-            fut = self._loop.create_future()
-
-            # This connection will now count towards the limit.
-            self._waiters[key].append(fut)
-
-            if traces:
-                for trace in traces:
-                    await trace.send_connection_queued_start()
-
-            try:
-                await fut
-            except BaseException as e:
-                if key in self._waiters:
-                    # remove a waiter even if it was cancelled, normally it's
-                    #  removed when it's notified
-                    try:
-                        self._waiters[key].remove(fut)
-                    except ValueError:  # fut may no longer be in list
-                        pass
-
-                raise e
-            finally:
-                if key in self._waiters and not self._waiters[key]:
-                    del self._waiters[key]
+        if (conn := await self._get(key, traces)) is not None:
+            # If we do not have to wait and we can get a connection from the pool
+            # we can avoid the timeout ceil logic and directly return the connection
+            return conn
 
-            if traces:
-                for trace in traces:
-                    await trace.send_connection_queued_end()
+        async with ceil_timeout(timeout.connect, timeout.ceil_threshold):
+            if self._available_connections(key) <= 0:
+                await self._wait_for_available_connection(key, traces)
+                if (conn := await self._get(key, traces)) is not None:
+                    return conn
 
-        proto = self._get(key)
-        if proto is None:
             placeholder = cast(ResponseHandler, _TransportPlaceholder())
             self._acquired.add(placeholder)
             self._acquired_per_host[key].add(placeholder)
 
-            if traces:
-                for trace in traces:
-                    await trace.send_connection_create_start()
-
             try:
+                # Traces are done inside the try block to ensure that the
+                # that the placeholder is still cleaned up if an exception
+                # is raised.
+                if traces:
+                    for trace in traces:
+                        await trace.send_connection_create_start()
                 proto = await self._create_connection(req, traces, timeout)
-                if self._closed:
-                    proto.close()
-                    raise ClientConnectionError("Connector is closed.")
+                if traces:
+                    for trace in traces:
+                        await trace.send_connection_create_end()
             except BaseException:
-                if not self._closed:
-                    self._acquired.remove(placeholder)
-                    self._drop_acquired_per_host(key, placeholder)
-                    self._release_waiter()
+                self._release_acquired(key, placeholder)
                 raise
             else:
-                if not self._closed:
-                    self._acquired.remove(placeholder)
-                    self._drop_acquired_per_host(key, placeholder)
-
-            if traces:
-                for trace in traces:
-                    await trace.send_connection_create_end()
-        else:
-            if traces:
-                # Acquire the connection to prevent race conditions with limits
-                placeholder = cast(ResponseHandler, _TransportPlaceholder())
-                self._acquired.add(placeholder)
-                self._acquired_per_host[key].add(placeholder)
-                for trace in traces:
-                    await trace.send_connection_reuseconn()
-                self._acquired.remove(placeholder)
-                self._drop_acquired_per_host(key, placeholder)
+                if self._closed:
+                    proto.close()
+                    raise ClientConnectionError("Connector is closed.")
 
+        # The connection was successfully created, drop the placeholder
+        # and add the real connection to the acquired set. There should
+        # be no awaits after the proto is added to the acquired set
+        # to ensure that the connection is not left in the acquired set
+        # on cancellation.
+        acquired_per_host = self._acquired_per_host[key]
+        self._acquired.remove(placeholder)
+        acquired_per_host.remove(placeholder)
         self._acquired.add(proto)
-        self._acquired_per_host[key].add(proto)
+        acquired_per_host.add(proto)
         return Connection(self, key, proto, self._loop)
 
-    def _get(self, key: "ConnectionKey") -> Optional[ResponseHandler]:
+    async def _wait_for_available_connection(
+        self, key: "ConnectionKey", traces: List["Trace"]
+    ) -> None:
+        """Wait for an available connection slot."""
+        # We loop here because there is a race between
+        # the connection limit check and the connection
+        # being acquired. If the connection is acquired
+        # between the check and the await statement, we
+        # need to loop again to check if the connection
+        # slot is still available.
+        attempts = 0
+        while True:
+            fut: asyncio.Future[None] = self._loop.create_future()
+            keyed_waiters = self._waiters[key]
+            keyed_waiters[fut] = None
+            if attempts:
+                # If we have waited before, we need to move the waiter
+                # to the front of the queue as otherwise we might get
+                # starved and hit the timeout.
+                keyed_waiters.move_to_end(fut, last=False)
+
+            try:
+                # Traces happen in the try block to ensure that the
+                # the waiter is still cleaned up if an exception is raised.
+                if traces:
+                    for trace in traces:
+                        await trace.send_connection_queued_start()
+                await fut
+                if traces:
+                    for trace in traces:
+                        await trace.send_connection_queued_end()
+            finally:
+                # pop the waiter from the queue if its still
+                # there and not already removed by _release_waiter
+                keyed_waiters.pop(fut, None)
+                if not self._waiters.get(key, True):
+                    del self._waiters[key]
+
+            if self._available_connections(key) > 0:
+                break
+            attempts += 1
+
+    async def _get(
+        self, key: "ConnectionKey", traces: List["Trace"]
+    ) -> Optional[Connection]:
+        """Get next reusable connection for the key or None.
+
+        The connection will be marked as acquired.
+        """
         try:
             conns = self._conns[key]
         except KeyError:
             return None
 
-        t1 = self._loop.time()
+        t1 = monotonic()
         while conns:
             proto, t0 = conns.pop()
-            if proto.is_connected():
-                if t1 - t0 > self._keepalive_timeout:
-                    transport = proto.transport
-                    proto.close()
-                    # only for SSL transports
-                    if key.is_ssl and not self._cleanup_closed_disabled:
-                        self._cleanup_closed_transports.append(transport)
-                else:
-                    if not conns:
-                        # The very last connection was reclaimed: drop the key
-                        del self._conns[key]
-                    return proto
-            else:
-                transport = proto.transport
-                proto.close()
-                if key.is_ssl and not self._cleanup_closed_disabled:
-                    self._cleanup_closed_transports.append(transport)
+            # We will we reuse the connection if its connected and
+            # the keepalive timeout has not been exceeded
+            if proto.is_connected() and t1 - t0 <= self._keepalive_timeout:
+                if not conns:
+                    # The very last connection was reclaimed: drop the key
+                    del self._conns[key]
+                self._acquired.add(proto)
+                self._acquired_per_host[key].add(proto)
+                if traces:
+                    for trace in traces:
+                        try:
+                            await trace.send_connection_reuseconn()
+                        except BaseException:
+                            self._release_acquired(key, proto)
+                            raise
+                return Connection(self, key, proto, self._loop)
+
+            # Connection cannot be reused, close it
+            transport = proto.transport
+            proto.close()
+            # only for SSL transports
+            if not self._cleanup_closed_disabled and key.is_ssl:
+                self._cleanup_closed_transports.append(transport)
 
         # No more connections: drop the key
         del self._conns[key]
@@ -630,7 +656,7 @@ class BaseConnector:
 
         # Having the dict keys ordered this avoids to iterate
         # at the same order at each call.
-        queues = list(self._waiters.keys())
+        queues = list(self._waiters)
         random.shuffle(queues)
 
         for key in queues:
@@ -639,25 +665,23 @@ class BaseConnector:
 
             waiters = self._waiters[key]
             while waiters:
-                waiter = waiters.popleft()
+                waiter, _ = waiters.popitem(last=False)
                 if not waiter.done():
                     waiter.set_result(None)
                     return
 
     def _release_acquired(self, key: "ConnectionKey", proto: ResponseHandler) -> None:
+        """Release acquired connection."""
         if self._closed:
             # acquired connection is already released on connector closing
             return
 
-        try:
-            self._acquired.remove(proto)
-            self._drop_acquired_per_host(key, proto)
-        except KeyError:  # pragma: no cover
-            # this may be result of undetermenistic order of objects
-            # finalization due garbage collection.
-            pass
-        else:
-            self._release_waiter()
+        self._acquired.discard(proto)
+        if conns := self._acquired_per_host.get(key):
+            conns.discard(proto)
+            if not conns:
+                del self._acquired_per_host[key]
+        self._release_waiter()
 
     def _release(
         self,
@@ -681,20 +705,21 @@ class BaseConnector:
 
             if key.is_ssl and not self._cleanup_closed_disabled:
                 self._cleanup_closed_transports.append(transport)
-        else:
-            conns = self._conns.get(key)
-            if conns is None:
-                conns = self._conns[key] = []
-            conns.append((protocol, self._loop.time()))
+            return
 
-            if self._cleanup_handle is None:
-                self._cleanup_handle = helpers.weakref_handle(
-                    self,
-                    "_cleanup",
-                    self._keepalive_timeout,
-                    self._loop,
-                    timeout_ceil_threshold=self._timeout_ceil_threshold,
-                )
+        conns = self._conns.get(key)
+        if conns is None:
+            conns = self._conns[key] = []
+        conns.append((protocol, monotonic()))
+
+        if self._cleanup_handle is None:
+            self._cleanup_handle = helpers.weakref_handle(
+                self,
+                "_cleanup",
+                self._keepalive_timeout,
+                self._loop,
+                timeout_ceil_threshold=self._timeout_ceil_threshold,
+            )
 
     async def _create_connection(
         self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
@@ -840,7 +865,9 @@ class TCPConnector(BaseConnector):
 
         self._use_dns_cache = use_dns_cache
         self._cached_hosts = _DNSCacheTable(ttl=ttl_dns_cache)
-        self._throttle_dns_events: Dict[Tuple[str, int], EventResultOrError] = {}
+        self._throttle_dns_futures: Dict[
+            Tuple[str, int], Set["asyncio.Future[None]"]
+        ] = {}
         self._family = family
         self._local_addr_infos = aiohappyeyeballs.addr_to_addr_infos(local_addr)
         self._happy_eyeballs_delay = happy_eyeballs_delay
@@ -849,8 +876,8 @@ class TCPConnector(BaseConnector):
 
     def close(self) -> Awaitable[None]:
         """Close all ongoing DNS calls."""
-        for ev in self._throttle_dns_events.values():
-            ev.cancel()
+        for fut in chain.from_iterable(self._throttle_dns_futures.values()):
+            fut.cancel()
 
         for t in self._resolve_host_tasks:
             t.cancel()
@@ -918,22 +945,46 @@ class TCPConnector(BaseConnector):
                     await trace.send_dns_cache_hit(host)
             return result
 
+        futures: Set["asyncio.Future[None]"]
         #
         # If multiple connectors are resolving the same host, we wait
         # for the first one to resolve and then use the result for all of them.
-        # We use a throttle event to ensure that we only resolve the host once
+        # We use a throttle to ensure that we only resolve the host once
         # and then use the result for all the waiters.
         #
+        if key in self._throttle_dns_futures:
+            # get futures early, before any await (#4014)
+            futures = self._throttle_dns_futures[key]
+            future: asyncio.Future[None] = self._loop.create_future()
+            futures.add(future)
+            if traces:
+                for trace in traces:
+                    await trace.send_dns_cache_hit(host)
+            try:
+                await future
+            finally:
+                futures.discard(future)
+            return self._cached_hosts.next_addrs(key)
+
+        # update dict early, before any await (#4014)
+        self._throttle_dns_futures[key] = futures = set()
         # In this case we need to create a task to ensure that we can shield
         # the task from cancellation as cancelling this lookup should not cancel
         # the underlying lookup or else the cancel event will get broadcast to
         # all the waiters across all connections.
         #
-        resolved_host_task = asyncio.create_task(
-            self._resolve_host_with_throttle(key, host, port, traces)
-        )
-        self._resolve_host_tasks.add(resolved_host_task)
-        resolved_host_task.add_done_callback(self._resolve_host_tasks.discard)
+        coro = self._resolve_host_with_throttle(key, host, port, futures, traces)
+        loop = asyncio.get_running_loop()
+        if sys.version_info >= (3, 12):
+            # Optimization for Python 3.12, try to send immediately
+            resolved_host_task = asyncio.Task(coro, loop=loop, eager_start=True)
+        else:
+            resolved_host_task = loop.create_task(coro)
+
+        if not resolved_host_task.done():
+            self._resolve_host_tasks.add(resolved_host_task)
+            resolved_host_task.add_done_callback(self._resolve_host_tasks.discard)
+
         try:
             return await asyncio.shield(resolved_host_task)
         except asyncio.CancelledError:
@@ -950,42 +1001,39 @@ class TCPConnector(BaseConnector):
         key: Tuple[str, int],
         host: str,
         port: int,
+        futures: Set["asyncio.Future[None]"],
         traces: Optional[Sequence["Trace"]],
     ) -> List[ResolveResult]:
-        """Resolve host with a dns events throttle."""
-        if key in self._throttle_dns_events:
-            # get event early, before any await (#4014)
-            event = self._throttle_dns_events[key]
+        """Resolve host and set result for all waiters.
+
+        This method must be run in a task and shielded from cancellation
+        to avoid cancelling the underlying lookup.
+        """
+        if traces:
+            for trace in traces:
+                await trace.send_dns_cache_miss(host)
+        try:
             if traces:
                 for trace in traces:
-                    await trace.send_dns_cache_hit(host)
-            await event.wait()
-        else:
-            # update dict early, before any await (#4014)
-            self._throttle_dns_events[key] = EventResultOrError(self._loop)
+                    await trace.send_dns_resolvehost_start(host)
+
+            addrs = await self._resolver.resolve(host, port, family=self._family)
             if traces:
                 for trace in traces:
-                    await trace.send_dns_cache_miss(host)
-            try:
-
-                if traces:
-                    for trace in traces:
-                        await trace.send_dns_resolvehost_start(host)
-
-                addrs = await self._resolver.resolve(host, port, family=self._family)
-                if traces:
-                    for trace in traces:
-                        await trace.send_dns_resolvehost_end(host)
+                    await trace.send_dns_resolvehost_end(host)
 
-                self._cached_hosts.add(key, addrs)
-                self._throttle_dns_events[key].set()
-            except BaseException as e:
-                # any DNS exception, independently of the implementation
-                # is set for the waiters to raise the same exception.
-                self._throttle_dns_events[key].set(exc=e)
-                raise
-            finally:
-                self._throttle_dns_events.pop(key)
+            self._cached_hosts.add(key, addrs)
+            for fut in futures:
+                set_result(fut, None)
+        except BaseException as e:
+            # any DNS exception is set for the waiters to raise the same exception.
+            # This coro is always run in task that is shielded from cancellation so
+            # we should never be propagating cancellation here.
+            for fut in futures:
+                set_exception(fut, e)
+            raise
+        finally:
+            self._throttle_dns_futures.pop(key)
 
         return self._cached_hosts.next_addrs(key)
 
@@ -1290,7 +1338,7 @@ class TCPConnector(BaseConnector):
                 raise
             # in case of proxy it is not ClientProxyConnectionError
             # it is problem of resolving proxy ip itself
-            raise ClientConnectorError(req.connection_key, exc) from exc
+            raise ClientConnectorDNSError(req.connection_key, exc) from exc
 
         last_exc: Optional[Exception] = None
         addr_infos = self._convert_hosts_to_addr_infos(hosts)
@@ -1311,7 +1359,7 @@ class TCPConnector(BaseConnector):
                     req=req,
                     client_error=client_error,
                 )
-            except ClientConnectorError as exc:
+            except (ClientConnectorError, asyncio.TimeoutError) as exc:
                 last_exc = exc
                 aiohappyeyeballs.pop_addr_infos_interleave(addr_infos, self._interleave)
                 continue
@@ -1412,7 +1460,6 @@ class TCPConnector(BaseConnector):
                 raise
             else:
                 conn._protocol = None
-                conn._transport = None
                 try:
                     if resp.status != 200:
                         message = resp.reason
diff --git a/contrib/python/aiohttp/aiohttp/helpers.py b/contrib/python/aiohttp/aiohttp/helpers.py
index 40705b16d71..6ee70786cfb 100644
--- a/contrib/python/aiohttp/aiohttp/helpers.py
+++ b/contrib/python/aiohttp/aiohttp/helpers.py
@@ -163,7 +163,9 @@ class BasicAuth(namedtuple("BasicAuth", ["login", "password", "encoding"])):
         """Create BasicAuth from url."""
         if not isinstance(url, URL):
             raise TypeError("url should be yarl.URL instance")
-        if url.user is None and url.password is None:
+        # Check raw_user and raw_password first as yarl is likely
+        # to already have these values parsed from the netloc in the cache.
+        if url.raw_user is None and url.raw_password is None:
             return None
         return cls(url.user or "", url.password or "", encoding=encoding)
 
@@ -174,11 +176,12 @@ class BasicAuth(namedtuple("BasicAuth", ["login", "password", "encoding"])):
 
 
 def strip_auth_from_url(url: URL) -> Tuple[URL, Optional[BasicAuth]]:
-    auth = BasicAuth.from_url(url)
-    if auth is None:
+    """Remove user and password from URL if present and return BasicAuth object."""
+    # Check raw_user and raw_password first as yarl is likely
+    # to already have these values parsed from the netloc in the cache.
+    if url.raw_user is None and url.raw_password is None:
         return url, None
-    else:
-        return url.with_user(None), auth
+    return url.with_user(None), BasicAuth(url.user or "", url.password or "")
 
 
 def netrc_from_env() -> Optional[netrc.netrc]:
@@ -614,6 +617,8 @@ def calculate_timeout_when(
 class TimeoutHandle:
     """Timeout handle"""
 
+    __slots__ = ("_timeout", "_loop", "_ceil_threshold", "_callbacks")
+
     def __init__(
         self,
         loop: asyncio.AbstractEventLoop,
@@ -662,11 +667,17 @@ class TimeoutHandle:
 
 
 class BaseTimerContext(ContextManager["BaseTimerContext"]):
+
+    __slots__ = ()
+
     def assert_timeout(self) -> None:
         """Raise TimeoutError if timeout has been exceeded."""
 
 
 class TimerNoop(BaseTimerContext):
+
+    __slots__ = ()
+
     def __enter__(self) -> BaseTimerContext:
         return self
 
@@ -682,10 +693,13 @@ class TimerNoop(BaseTimerContext):
 class TimerContext(BaseTimerContext):
     """Low resolution timeout context manager"""
 
+    __slots__ = ("_loop", "_tasks", "_cancelled", "_cancelling")
+
     def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
         self._loop = loop
         self._tasks: List[asyncio.Task[Any]] = []
         self._cancelled = False
+        self._cancelling = 0
 
     def assert_timeout(self) -> None:
         """Raise TimeoutError if timer has already been cancelled."""
@@ -694,12 +708,17 @@ class TimerContext(BaseTimerContext):
 
     def __enter__(self) -> BaseTimerContext:
         task = asyncio.current_task(loop=self._loop)
-
         if task is None:
             raise RuntimeError(
                 "Timeout context manager should be used " "inside a task"
             )
 
+        if sys.version_info >= (3, 11):
+            # Remember if the task was already cancelling
+            # so when we __exit__ we can decide if we should
+            # raise asyncio.TimeoutError or let the cancellation propagate
+            self._cancelling = task.cancelling()
+
         if self._cancelled:
             raise asyncio.TimeoutError from None
 
@@ -712,11 +731,22 @@ class TimerContext(BaseTimerContext):
         exc_val: Optional[BaseException],
         exc_tb: Optional[TracebackType],
     ) -> Optional[bool]:
+        enter_task: Optional[asyncio.Task[Any]] = None
         if self._tasks:
-            self._tasks.pop()
+            enter_task = self._tasks.pop()
 
         if exc_type is asyncio.CancelledError and self._cancelled:
-            raise asyncio.TimeoutError from None
+            assert enter_task is not None
+            # The timeout was hit, and the task was cancelled
+            # so we need to uncancel the last task that entered the context manager
+            # since the cancellation should not leak out of the context manager
+            if sys.version_info >= (3, 11):
+                # If the task was already cancelling don't raise
+                # asyncio.TimeoutError and instead return None
+                # to allow the cancellation to propagate
+                if enter_task.uncancel() > self._cancelling:
+                    return None
+            raise asyncio.TimeoutError from exc_val
         return None
 
     def timeout(self) -> None:
@@ -784,11 +814,7 @@ class HeadersMixin:
     def content_length(self) -> Optional[int]:
         """The value of Content-Length HTTP header."""
         content_length = self._headers.get(hdrs.CONTENT_LENGTH)
-
-        if content_length is not None:
-            return int(content_length)
-        else:
-            return None
+        return None if content_length is None else int(content_length)
 
 
 def set_result(fut: "asyncio.Future[_T]", result: _T) -> None:
diff --git a/contrib/python/aiohttp/aiohttp/http_parser.py b/contrib/python/aiohttp/aiohttp/http_parser.py
index 686a2d02e28..9fc7e8a2c8b 100644
--- a/contrib/python/aiohttp/aiohttp/http_parser.py
+++ b/contrib/python/aiohttp/aiohttp/http_parser.py
@@ -845,6 +845,13 @@ class HttpPayloadParser:
                         i = chunk.find(CHUNK_EXT, 0, pos)
                         if i >= 0:
                             size_b = chunk[:i]  # strip chunk-extensions
+                            # Verify no LF in the chunk-extension
+                            if b"\n" in (ext := chunk[i:pos]):
+                                exc = BadHttpMessage(
+                                    f"Unexpected LF in chunk-extension: {ext!r}"
+                                )
+                                set_exception(self.payload, exc)
+                                raise exc
                         else:
                             size_b = chunk[:pos]
 
diff --git a/contrib/python/aiohttp/aiohttp/http_websocket.py b/contrib/python/aiohttp/aiohttp/http_websocket.py
index fb00ebc7d35..0916530dbe8 100644
--- a/contrib/python/aiohttp/aiohttp/http_websocket.py
+++ b/contrib/python/aiohttp/aiohttp/http_websocket.py
@@ -133,8 +133,12 @@ class WSMessage(NamedTuple):
         return loads(self.data)
 
 
-WS_CLOSED_MESSAGE = WSMessage(WSMsgType.CLOSED, None, None)
-WS_CLOSING_MESSAGE = WSMessage(WSMsgType.CLOSING, None, None)
+# Constructing the tuple directly to avoid the overhead of
+# the lambda and arg processing since NamedTuples are constructed
+# with a run time built lambda
+# https://github.com/python/cpython/blob/d83fcf8371f2f33c7797bc8f5423a8bca8c46e5c/Lib/collections/__init__.py#L441
+WS_CLOSED_MESSAGE = tuple.__new__(WSMessage, (WSMsgType.CLOSED, None, None))
+WS_CLOSING_MESSAGE = tuple.__new__(WSMessage, (WSMsgType.CLOSING, None, None))
 
 
 class WebSocketError(Exception):
@@ -411,12 +415,14 @@ class WebSocketReader:
                             WSCloseCode.INVALID_TEXT, "Invalid UTF-8 text message"
                         ) from exc
 
-                    self.queue.feed_data(WSMessage(WSMsgType.TEXT, text, ""), len(text))
+                    # tuple.__new__ is used to avoid the overhead of the lambda
+                    msg = tuple.__new__(WSMessage, (WSMsgType.TEXT, text, ""))
+                    self.queue.feed_data(msg, len(payload_merged))
                     continue
 
-                self.queue.feed_data(
-                    WSMessage(WSMsgType.BINARY, payload_merged, ""), len(payload_merged)
-                )
+                # tuple.__new__ is used to avoid the overhead of the lambda
+                msg = tuple.__new__(WSMessage, (WSMsgType.BINARY, payload_merged, ""))
+                self.queue.feed_data(msg, len(payload_merged))
             elif opcode == WSMsgType.CLOSE:
                 if len(payload) >= 2:
                     close_code = UNPACK_CLOSE_CODE(payload[:2])[0]
@@ -431,26 +437,26 @@ class WebSocketReader:
                         raise WebSocketError(
                             WSCloseCode.INVALID_TEXT, "Invalid UTF-8 text message"
                         ) from exc
-                    msg = WSMessage(WSMsgType.CLOSE, close_code, close_message)
+                    msg = tuple.__new__(
+                        WSMessage, (WSMsgType.CLOSE, close_code, close_message)
+                    )
                 elif payload:
                     raise WebSocketError(
                         WSCloseCode.PROTOCOL_ERROR,
                         f"Invalid close frame: {fin} {opcode} {payload!r}",
                     )
                 else:
-                    msg = WSMessage(WSMsgType.CLOSE, 0, "")
+                    msg = tuple.__new__(WSMessage, (WSMsgType.CLOSE, 0, ""))
 
                 self.queue.feed_data(msg, 0)
 
             elif opcode == WSMsgType.PING:
-                self.queue.feed_data(
-                    WSMessage(WSMsgType.PING, payload, ""), len(payload)
-                )
+                msg = tuple.__new__(WSMessage, (WSMsgType.PING, payload, ""))
+                self.queue.feed_data(msg, len(payload))
 
             elif opcode == WSMsgType.PONG:
-                self.queue.feed_data(
-                    WSMessage(WSMsgType.PONG, payload, ""), len(payload)
-                )
+                msg = tuple.__new__(WSMessage, (WSMsgType.PONG, payload, ""))
+                self.queue.feed_data(msg, len(payload))
 
             else:
                 raise WebSocketError(
@@ -669,13 +675,16 @@ class WebSocketWriter:
         if msg_length < 126:
             header = PACK_LEN1(first_byte, msg_length | mask_bit)
             header_len = 2
-        elif msg_length < (1 << 16):
+        elif msg_length < 65536:
             header = PACK_LEN2(first_byte, 126 | mask_bit, msg_length)
             header_len = 4
         else:
             header = PACK_LEN3(first_byte, 127 | mask_bit, msg_length)
             header_len = 10
 
+        if self.transport.is_closing():
+            raise ClientConnectionResetError("Cannot write to closing transport")
+
         # https://datatracker.ietf.org/doc/html/rfc6455#section-5.3
         # If we are using a mask, we need to generate it randomly
         # and apply it to the message before sending it. A mask is
@@ -687,17 +696,15 @@ class WebSocketWriter:
             mask = PACK_RANDBITS(self.get_random_bits())
             message = bytearray(message)
             _websocket_mask(mask, message)
-            self._write(header + mask + message)
-            self._output_size += header_len + MASK_LEN + msg_length
-
+            self.transport.write(header + mask + message)
+            self._output_size += MASK_LEN
+        elif msg_length > MSG_SIZE:
+            self.transport.write(header)
+            self.transport.write(message)
         else:
-            if msg_length > MSG_SIZE:
-                self._write(header)
-                self._write(message)
-            else:
-                self._write(header + message)
+            self.transport.write(header + message)
 
-            self._output_size += header_len + msg_length
+        self._output_size += header_len + msg_length
 
         # It is safe to return control to the event loop when using compression
         # after this point as we have already sent or buffered all the data.
@@ -718,11 +725,6 @@ class WebSocketWriter:
             max_sync_chunk_size=WEBSOCKET_MAX_SYNC_CHUNK_SIZE,
         )
 
-    def _write(self, data: bytes) -> None:
-        if self.transport is None or self.transport.is_closing():
-            raise ClientConnectionResetError("Cannot write to closing transport")
-        self.transport.write(data)
-
     async def pong(self, message: Union[bytes, str] = b"") -> None:
         """Send pong message."""
         if isinstance(message, str):
diff --git a/contrib/python/aiohttp/aiohttp/locks.py b/contrib/python/aiohttp/aiohttp/locks.py
deleted file mode 100644
index de2dc83d09d..00000000000
--- a/contrib/python/aiohttp/aiohttp/locks.py
+++ /dev/null
@@ -1,41 +0,0 @@
-import asyncio
-import collections
-from typing import Any, Deque, Optional
-
-
-class EventResultOrError:
-    """Event asyncio lock helper class.
-
-    Wraps the Event asyncio lock allowing either to awake the
-    locked Tasks without any error or raising an exception.
-
-    thanks to @vorpalsmith for the simple design.
-    """
-
-    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
-        self._loop = loop
-        self._exc: Optional[BaseException] = None
-        self._event = asyncio.Event()
-        self._waiters: Deque[asyncio.Future[Any]] = collections.deque()
-
-    def set(self, exc: Optional[BaseException] = None) -> None:
-        self._exc = exc
-        self._event.set()
-
-    async def wait(self) -> Any:
-        waiter = self._loop.create_task(self._event.wait())
-        self._waiters.append(waiter)
-        try:
-            val = await waiter
-        finally:
-            self._waiters.remove(waiter)
-
-        if self._exc is not None:
-            raise self._exc
-
-        return val
-
-    def cancel(self) -> None:
-        """Cancel all waiters"""
-        for waiter in self._waiters:
-            waiter.cancel()
diff --git a/contrib/python/aiohttp/aiohttp/payload.py b/contrib/python/aiohttp/aiohttp/payload.py
index 27636977774..151f9dd497b 100644
--- a/contrib/python/aiohttp/aiohttp/payload.py
+++ b/contrib/python/aiohttp/aiohttp/payload.py
@@ -101,6 +101,7 @@ class PayloadRegistry:
         self._first: List[_PayloadRegistryItem] = []
         self._normal: List[_PayloadRegistryItem] = []
         self._last: List[_PayloadRegistryItem] = []
+        self._normal_lookup: Dict[Any, PayloadType] = {}
 
     def get(
         self,
@@ -109,12 +110,20 @@ class PayloadRegistry:
         _CHAIN: "Type[chain[_PayloadRegistryItem]]" = chain,
         **kwargs: Any,
     ) -> "Payload":
+        if self._first:
+            for factory, type_ in self._first:
+                if isinstance(data, type_):
+                    return factory(data, *args, **kwargs)
+        # Try the fast lookup first
+        if lookup_factory := self._normal_lookup.get(type(data)):
+            return lookup_factory(data, *args, **kwargs)
+        # Bail early if its already a Payload
         if isinstance(data, Payload):
             return data
-        for factory, type in _CHAIN(self._first, self._normal, self._last):
-            if isinstance(data, type):
+        # Fallback to the slower linear search
+        for factory, type_ in _CHAIN(self._normal, self._last):
+            if isinstance(data, type_):
                 return factory(data, *args, **kwargs)
-
         raise LookupError()
 
     def register(
@@ -124,6 +133,11 @@ class PayloadRegistry:
             self._first.append((factory, type))
         elif order is Order.normal:
             self._normal.append((factory, type))
+            if isinstance(type, Iterable):
+                for t in type:
+                    self._normal_lookup[t] = factory
+            else:
+                self._normal_lookup[type] = factory
         elif order is Order.try_last:
             self._last.append((factory, type))
         else:
@@ -159,7 +173,8 @@ class Payload(ABC):
             self._headers[hdrs.CONTENT_TYPE] = content_type
         else:
             self._headers[hdrs.CONTENT_TYPE] = self._default_content_type
-        self._headers.update(headers or {})
+        if headers:
+            self._headers.update(headers)
 
     @property
     def size(self) -> Optional[int]:
@@ -228,9 +243,6 @@ class BytesPayload(Payload):
     def __init__(
         self, value: Union[bytes, bytearray, memoryview], *args: Any, **kwargs: Any
     ) -> None:
-        if not isinstance(value, (bytes, bytearray, memoryview)):
-            raise TypeError(f"value argument must be byte-ish, not {type(value)!r}")
-
         if "content_type" not in kwargs:
             kwargs["content_type"] = "application/octet-stream"
 
@@ -238,8 +250,10 @@ class BytesPayload(Payload):
 
         if isinstance(value, memoryview):
             self._size = value.nbytes
-        else:
+        elif isinstance(value, (bytes, bytearray)):
             self._size = len(value)
+        else:
+            raise TypeError(f"value argument must be byte-ish, not {type(value)!r}")
 
         if self._size > TOO_LARGE_BYTES_BODY:
             kwargs = {"source": self}
diff --git a/contrib/python/aiohttp/aiohttp/resolver.py b/contrib/python/aiohttp/aiohttp/resolver.py
index 06855fa13fd..5bcdb8ada43 100644
--- a/contrib/python/aiohttp/aiohttp/resolver.py
+++ b/contrib/python/aiohttp/aiohttp/resolver.py
@@ -111,7 +111,7 @@ class AsyncResolver(AbstractResolver):
             )
         except aiodns.error.DNSError as exc:
             msg = exc.args[1] if len(exc.args) >= 1 else "DNS lookup failed"
-            raise OSError(msg) from exc
+            raise OSError(None, msg) from exc
         hosts: List[ResolveResult] = []
         for node in resp.nodes:
             address: Union[Tuple[bytes, int], Tuple[bytes, int, int, int]] = node.addr
@@ -145,7 +145,7 @@ class AsyncResolver(AbstractResolver):
             )
 
         if not hosts:
-            raise OSError("DNS lookup failed")
+            raise OSError(None, "DNS lookup failed")
 
         return hosts
 
@@ -161,7 +161,7 @@ class AsyncResolver(AbstractResolver):
             resp = await self._resolver.query(host, qtype)
         except aiodns.error.DNSError as exc:
             msg = exc.args[1] if len(exc.args) >= 1 else "DNS lookup failed"
-            raise OSError(msg) from exc
+            raise OSError(None, msg) from exc
 
         hosts = []
         for rr in resp:
@@ -177,7 +177,7 @@ class AsyncResolver(AbstractResolver):
             )
 
         if not hosts:
-            raise OSError("DNS lookup failed")
+            raise OSError(None, "DNS lookup failed")
 
         return hosts
 
diff --git a/contrib/python/aiohttp/aiohttp/streams.py b/contrib/python/aiohttp/aiohttp/streams.py
index c927cfbb1b3..030c6735947 100644
--- a/contrib/python/aiohttp/aiohttp/streams.py
+++ b/contrib/python/aiohttp/aiohttp/streams.py
@@ -245,9 +245,10 @@ class StreamReader(AsyncStreamReaderMixin):
         if not data:
             return
 
-        self._size += len(data)
+        data_len = len(data)
+        self._size += data_len
         self._buffer.append(data)
-        self.total_bytes += len(data)
+        self.total_bytes += data_len
 
         waiter = self._waiter
         if waiter is not None:
diff --git a/contrib/python/aiohttp/aiohttp/test_utils.py b/contrib/python/aiohttp/aiohttp/test_utils.py
index 01496b6711a..850efcb8b65 100644
--- a/contrib/python/aiohttp/aiohttp/test_utils.py
+++ b/contrib/python/aiohttp/aiohttp/test_utils.py
@@ -148,7 +148,7 @@ class BaseTestServer(ABC):
         assert self._root is not None
         url = URL(path)
         if not self.skip_url_asserts:
-            assert not url.is_absolute()
+            assert not url.absolute
             return self._root.join(url)
         else:
             return URL(str(self._root) + str(path))
diff --git a/contrib/python/aiohttp/aiohttp/tracing.py b/contrib/python/aiohttp/aiohttp/tracing.py
index a42fc0d2a44..067a132464e 100644
--- a/contrib/python/aiohttp/aiohttp/tracing.py
+++ b/contrib/python/aiohttp/aiohttp/tracing.py
@@ -1,5 +1,5 @@
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any, Awaitable, Mapping, Optional, Protocol, Type, TypeVar
+from typing import TYPE_CHECKING, Awaitable, Mapping, Optional, Protocol, Type, TypeVar
 
 import attr
 from aiosignal import Signal
@@ -64,7 +64,7 @@ class TraceConfig:
         self._trace_config_ctx_factory = trace_config_ctx_factory
 
     def trace_config_ctx(
-        self, trace_request_ctx: Optional[Mapping[str, Any]] = None
+        self, trace_request_ctx: Optional[Mapping[str, str]] = None
     ) -> SimpleNamespace:
         """Return a new trace_config_ctx instance"""
         return self._trace_config_ctx_factory(trace_request_ctx=trace_request_ctx)
diff --git a/contrib/python/aiohttp/aiohttp/typedefs.py b/contrib/python/aiohttp/aiohttp/typedefs.py
index 2e285fa2561..cc8c0825b4e 100644
--- a/contrib/python/aiohttp/aiohttp/typedefs.py
+++ b/contrib/python/aiohttp/aiohttp/typedefs.py
@@ -8,23 +8,12 @@ from typing import (
     Iterable,
     Mapping,
     Protocol,
-    Sequence,
     Tuple,
     Union,
 )
 
 from multidict import CIMultiDict, CIMultiDictProxy, MultiDict, MultiDictProxy, istr
-from yarl import URL
-
-try:
-    # Available in yarl>=1.10.0
-    from yarl import Query as _Query
-except ImportError:  # pragma: no cover
-    SimpleQuery = Union[str, int, float]  # pragma: no cover
-    QueryVariable = Union[SimpleQuery, "Sequence[SimpleQuery]"]  # pragma: no cover
-    _Query = Union[  # type: ignore[misc]  # pragma: no cover
-        None, str, "Mapping[str, QueryVariable]", "Sequence[Tuple[str, QueryVariable]]"
-    ]
+from yarl import URL, Query as _Query
 
 Query = _Query
 
diff --git a/contrib/python/aiohttp/aiohttp/web_app.py b/contrib/python/aiohttp/aiohttp/web_app.py
index ab11981a8a5..628e098d2ea 100644
--- a/contrib/python/aiohttp/aiohttp/web_app.py
+++ b/contrib/python/aiohttp/aiohttp/web_app.py
@@ -54,6 +54,7 @@ from .web_urldispatcher import (
     MaskDomain,
     MatchedSubAppResource,
     PrefixedSubAppResource,
+    SystemRoute,
     UrlDispatcher,
 )
 
@@ -79,7 +80,6 @@ _U = TypeVar("_U")
 _Resource = TypeVar("_Resource", bound=AbstractResource)
 
 
-@lru_cache(None)
 def _build_middlewares(
     handler: Handler, apps: Tuple["Application", ...]
 ) -> Callable[[Request], Awaitable[StreamResponse]]:
@@ -90,6 +90,9 @@ def _build_middlewares(
     return handler
 
 
+_cached_build_middleware = lru_cache(maxsize=1024)(_build_middlewares)
+
+
 class Application(MutableMapping[Union[str, AppKey[Any]], Any]):
     ATTRS = frozenset(
         [
@@ -544,8 +547,13 @@ class Application(MutableMapping[Union[str, AppKey[Any]], Any]):
         handler = match_info.handler
 
         if self._run_middlewares:
-            if not self._has_legacy_middlewares:
-                handler = _build_middlewares(handler, match_info.apps)
+            # If its a SystemRoute, don't cache building the middlewares since
+            # they are constructed for every MatchInfoError as a new handler
+            # is made each time.
+            if not self._has_legacy_middlewares and not isinstance(
+                match_info.route, SystemRoute
+            ):
+                handler = _cached_build_middleware(handler, match_info.apps)
             else:
                 for app in match_info.apps[::-1]:
                     for m, new_style in app._middlewares_handlers:  # type: ignore[union-attr]
diff --git a/contrib/python/aiohttp/aiohttp/web_request.py b/contrib/python/aiohttp/aiohttp/web_request.py
index eca71e4413a..62a08ea248b 100644
--- a/contrib/python/aiohttp/aiohttp/web_request.py
+++ b/contrib/python/aiohttp/aiohttp/web_request.py
@@ -174,7 +174,7 @@ class BaseRequest(MutableMapping[str, Any], HeadersMixin):
         self._version = message.version
         self._cache: Dict[str, Any] = {}
         url = message.url
-        if url.is_absolute():
+        if url.absolute:
             if scheme is not None:
                 url = url.with_scheme(scheme)
             if host is not None:
@@ -431,6 +431,10 @@ class BaseRequest(MutableMapping[str, Any], HeadersMixin):
         - overridden value by .clone(host=new_host) call.
         - HOST HTTP header
         - socket.getfqdn() value
+
+        For example, 'example.com' or 'localhost:8080'.
+
+        For historical reasons, the port number may be included.
         """
         host = self._message.headers.get(hdrs.HOST)
         if host is not None:
@@ -454,8 +458,10 @@ class BaseRequest(MutableMapping[str, Any], HeadersMixin):
 
     @reify
     def url(self) -> URL:
-        url = URL.build(scheme=self.scheme, host=self.host)
-        return url.join(self._rel_url)
+        """The full URL of the request."""
+        # authority is used here because it may include the port number
+        # and we want yarl to parse it correctly
+        return URL.build(scheme=self.scheme, authority=self.host).join(self._rel_url)
 
     @reify
     def path(self) -> str:
diff --git a/contrib/python/aiohttp/aiohttp/web_response.py b/contrib/python/aiohttp/aiohttp/web_response.py
index 956359718da..2db96222709 100644
--- a/contrib/python/aiohttp/aiohttp/web_response.py
+++ b/contrib/python/aiohttp/aiohttp/web_response.py
@@ -42,6 +42,7 @@ from .payload import Payload
 from .typedefs import JSONEncoder, LooseHeaders
 
 REASON_PHRASES = {http_status.value: http_status.phrase for http_status in HTTPStatus}
+LARGE_BODY_SIZE = 1024**2
 
 __all__ = ("ContentCoding", "StreamResponse", "Response", "json_response")
 
@@ -395,25 +396,26 @@ class StreamResponse(BaseClass, HeadersMixin):
         self._headers[CONTENT_TYPE] = ctype
 
     async def _do_start_compression(self, coding: ContentCoding) -> None:
-        if coding != ContentCoding.identity:
-            assert self._payload_writer is not None
-            self._headers[hdrs.CONTENT_ENCODING] = coding.value
-            self._payload_writer.enable_compression(coding.value)
-            # Compressed payload may have different content length,
-            # remove the header
-            self._headers.popall(hdrs.CONTENT_LENGTH, None)
+        if coding is ContentCoding.identity:
+            return
+        assert self._payload_writer is not None
+        self._headers[hdrs.CONTENT_ENCODING] = coding.value
+        self._payload_writer.enable_compression(coding.value)
+        # Compressed payload may have different content length,
+        # remove the header
+        self._headers.popall(hdrs.CONTENT_LENGTH, None)
 
     async def _start_compression(self, request: "BaseRequest") -> None:
         if self._compression_force:
             await self._do_start_compression(self._compression_force)
-        else:
-            # Encoding comparisons should be case-insensitive
-            # https://www.rfc-editor.org/rfc/rfc9110#section-8.4.1
-            accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()
-            for value, coding in CONTENT_CODINGS.items():
-                if value in accept_encoding:
-                    await self._do_start_compression(coding)
-                    return
+            return
+        # Encoding comparisons should be case-insensitive
+        # https://www.rfc-editor.org/rfc/rfc9110#section-8.4.1
+        accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()
+        for value, coding in CONTENT_CODINGS.items():
+            if value in accept_encoding:
+                await self._do_start_compression(coding)
+                return
 
     async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter]:
         if self._eof_sent:
@@ -446,9 +448,10 @@ class StreamResponse(BaseClass, HeadersMixin):
         version = request.version
 
         headers = self._headers
-        for cookie in self._cookies.values():
-            value = cookie.output(header="")[1:]
-            headers.add(hdrs.SET_COOKIE, value)
+        if self._cookies:
+            for cookie in self._cookies.values():
+                value = cookie.output(header="")[1:]
+                headers.add(hdrs.SET_COOKIE, value)
 
         if self._compression:
             await self._start_compression(request)
@@ -464,7 +467,7 @@ class StreamResponse(BaseClass, HeadersMixin):
                 headers[hdrs.TRANSFER_ENCODING] = "chunked"
             if hdrs.CONTENT_LENGTH in headers:
                 del headers[hdrs.CONTENT_LENGTH]
-        elif self._length_check:
+        elif self._length_check:  # Disabled for WebSockets
             writer.length = self.content_length
             if writer.length is None:
                 if version >= HttpVersion11:
@@ -485,7 +488,7 @@ class StreamResponse(BaseClass, HeadersMixin):
             # https://datatracker.ietf.org/doc/html/rfc9112#section-6.1-13
             if hdrs.TRANSFER_ENCODING in headers:
                 del headers[hdrs.TRANSFER_ENCODING]
-        elif self.content_length != 0:
+        elif (writer.length if self._length_check else self.content_length) != 0:
             # https://www.rfc-editor.org/rfc/rfc9110#section-8.3-5
             headers.setdefault(hdrs.CONTENT_TYPE, "application/octet-stream")
         headers.setdefault(hdrs.DATE, rfc822_formatted_time())
@@ -496,9 +499,8 @@ class StreamResponse(BaseClass, HeadersMixin):
             if keep_alive:
                 if version == HttpVersion10:
                     headers[hdrs.CONNECTION] = "keep-alive"
-            else:
-                if version == HttpVersion11:
-                    headers[hdrs.CONNECTION] = "close"
+            elif version == HttpVersion11:
+                headers[hdrs.CONNECTION] = "close"
 
     async def _write_headers(self) -> None:
         request = self._req
@@ -626,19 +628,17 @@ class Response(StreamResponse):
                 real_headers[hdrs.CONTENT_TYPE] = content_type + "; charset=" + charset
                 body = text.encode(charset)
                 text = None
-        else:
-            if hdrs.CONTENT_TYPE in real_headers:
-                if content_type is not None or charset is not None:
-                    raise ValueError(
-                        "passing both Content-Type header and "
-                        "content_type or charset params "
-                        "is forbidden"
-                    )
-            else:
-                if content_type is not None:
-                    if charset is not None:
-                        content_type += "; charset=" + charset
-                    real_headers[hdrs.CONTENT_TYPE] = content_type
+        elif hdrs.CONTENT_TYPE in real_headers:
+            if content_type is not None or charset is not None:
+                raise ValueError(
+                    "passing both Content-Type header and "
+                    "content_type or charset params "
+                    "is forbidden"
+                )
+        elif content_type is not None:
+            if charset is not None:
+                content_type += "; charset=" + charset
+            real_headers[hdrs.CONTENT_TYPE] = content_type
 
         super().__init__(status=status, reason=reason, headers=real_headers)
 
@@ -707,7 +707,7 @@ class Response(StreamResponse):
             return None
 
         if hdrs.CONTENT_LENGTH in self._headers:
-            return super().content_length
+            return int(self._headers[hdrs.CONTENT_LENGTH])
 
         if self._compressed_body is not None:
             # Return length of the compressed body
@@ -734,16 +734,13 @@ class Response(StreamResponse):
         assert not data, f"data arg is not supported, got {data!r}"
         assert self._req is not None
         assert self._payload_writer is not None
-        if body is not None:
-            if self._must_be_empty_body:
-                await super().write_eof()
-            elif isinstance(self._body, Payload):
-                await self._body.write(self._payload_writer)
-                await super().write_eof()
-            else:
-                await super().write_eof(cast(bytes, body))
-        else:
+        if body is None or self._must_be_empty_body:
+            await super().write_eof()
+        elif isinstance(self._body, Payload):
+            await self._body.write(self._payload_writer)
             await super().write_eof()
+        else:
+            await super().write_eof(cast(bytes, body))
 
     async def _start(self, request: "BaseRequest") -> AbstractStreamWriter:
         if hdrs.CONTENT_LENGTH in self._headers:
@@ -766,30 +763,28 @@ class Response(StreamResponse):
     async def _do_start_compression(self, coding: ContentCoding) -> None:
         if self._chunked or isinstance(self._body, Payload):
             return await super()._do_start_compression(coding)
-
-        if coding != ContentCoding.identity:
-            # Instead of using _payload_writer.enable_compression,
-            # compress the whole body
-            compressor = ZLibCompressor(
-                encoding=str(coding.value),
-                max_sync_chunk_size=self._zlib_executor_size,
-                executor=self._zlib_executor,
-            )
-            assert self._body is not None
-            if self._zlib_executor_size is None and len(self._body) > 1024 * 1024:
-                warnings.warn(
-                    "Synchronous compression of large response bodies "
-                    f"({len(self._body)} bytes) might block the async event loop. "
-                    "Consider providing a custom value to zlib_executor_size/"
-                    "zlib_executor response properties or disabling compression on it."
-                )
-            self._compressed_body = (
-                await compressor.compress(self._body) + compressor.flush()
+        if coding is ContentCoding.identity:
+            return
+        # Instead of using _payload_writer.enable_compression,
+        # compress the whole body
+        compressor = ZLibCompressor(
+            encoding=coding.value,
+            max_sync_chunk_size=self._zlib_executor_size,
+            executor=self._zlib_executor,
+        )
+        assert self._body is not None
+        if self._zlib_executor_size is None and len(self._body) > LARGE_BODY_SIZE:
+            warnings.warn(
+                "Synchronous compression of large response bodies "
+                f"({len(self._body)} bytes) might block the async event loop. "
+                "Consider providing a custom value to zlib_executor_size/"
+                "zlib_executor response properties or disabling compression on it."
             )
-            assert self._compressed_body is not None
-
-            self._headers[hdrs.CONTENT_ENCODING] = coding.value
-            self._headers[hdrs.CONTENT_LENGTH] = str(len(self._compressed_body))
+        self._compressed_body = (
+            await compressor.compress(self._body) + compressor.flush()
+        )
+        self._headers[hdrs.CONTENT_ENCODING] = coding.value
+        self._headers[hdrs.CONTENT_LENGTH] = str(len(self._compressed_body))
 
 
 def json_response(
diff --git a/contrib/python/aiohttp/aiohttp/web_ws.py b/contrib/python/aiohttp/aiohttp/web_ws.py
index 382223097ea..0e8adef6f8d 100644
--- a/contrib/python/aiohttp/aiohttp/web_ws.py
+++ b/contrib/python/aiohttp/aiohttp/web_ws.py
@@ -379,14 +379,14 @@ class WebSocketResponse(StreamResponse):
             raise RuntimeError("Call .prepare() first")
         await self._writer.pong(message)
 
-    async def send_str(self, data: str, compress: Optional[bool] = None) -> None:
+    async def send_str(self, data: str, compress: Optional[int] = None) -> None:
         if self._writer is None:
             raise RuntimeError("Call .prepare() first")
         if not isinstance(data, str):
             raise TypeError("data argument must be str (%r)" % type(data))
         await self._writer.send(data, binary=False, compress=compress)
 
-    async def send_bytes(self, data: bytes, compress: Optional[bool] = None) -> None:
+    async def send_bytes(self, data: bytes, compress: Optional[int] = None) -> None:
         if self._writer is None:
             raise RuntimeError("Call .prepare() first")
         if not isinstance(data, (bytes, bytearray, memoryview)):
@@ -396,7 +396,7 @@ class WebSocketResponse(StreamResponse):
     async def send_json(
         self,
         data: Any,
-        compress: Optional[bool] = None,
+        compress: Optional[int] = None,
         *,
         dumps: JSONEncoder = json.dumps,
     ) -> None:
@@ -453,7 +453,11 @@ class WebSocketResponse(StreamResponse):
 
         try:
             async with async_timeout.timeout(self._timeout):
-                msg = await reader.read()
+                while True:
+                    msg = await reader.read()
+                    if msg.type is WSMsgType.CLOSE:
+                        self._set_code_close_transport(msg.data)
+                        return True
         except asyncio.CancelledError:
             self._set_code_close_transport(WSCloseCode.ABNORMAL_CLOSURE)
             raise
@@ -462,14 +466,6 @@ class WebSocketResponse(StreamResponse):
             self._set_code_close_transport(WSCloseCode.ABNORMAL_CLOSURE)
             return True
 
-        if msg.type is WSMsgType.CLOSE:
-            self._set_code_close_transport(msg.data)
-            return True
-
-        self._set_code_close_transport(WSCloseCode.ABNORMAL_CLOSURE)
-        self._exception = asyncio.TimeoutError()
-        return True
-
     def _set_closing(self, code: WSCloseCode) -> None:
         """Set the close code and mark the connection as closing."""
         self._closing = True
@@ -490,8 +486,6 @@ class WebSocketResponse(StreamResponse):
         if self._reader is None:
             raise RuntimeError("Call .prepare() first")
 
-        loop = self._loop
-        assert loop is not None
         receive_timeout = timeout or self._receive_timeout
         while True:
             if self._waiting:
diff --git a/contrib/python/aiohttp/patches/06-pr9398-fix-trace-request-ctx-typing.patch b/contrib/python/aiohttp/patches/06-pr9398-fix-trace-request-ctx-typing.patch
deleted file mode 100644
index 55b5b76f4d6..00000000000
--- a/contrib/python/aiohttp/patches/06-pr9398-fix-trace-request-ctx-typing.patch
+++ /dev/null
@@ -1,32 +0,0 @@
-#
-#    back-port pull-req #9398
-#
---- contrib/python/aiohttp/aiohttp/client.py
-+++ contrib/python/aiohttp/aiohttp/client.py
-@@ -178,7 +178,7 @@ class _RequestOptions(TypedDict, total=False):
-     ssl: Union[SSLContext, bool, Fingerprint]
-     server_hostname: Union[str, None]
-     proxy_headers: Union[LooseHeaders, None]
--    trace_request_ctx: Union[Mapping[str, str], None]
-+    trace_request_ctx: Union[Mapping[str, Any], None]
-     read_bufsize: Union[int, None]
-     auto_decompress: Union[bool, None]
-     max_line_size: Union[int, None]
---- contrib/python/aiohttp/aiohttp/tracing.py
-+++ contrib/python/aiohttp/aiohttp/tracing.py
-@@ -1,5 +1,5 @@
- from types import SimpleNamespace
--from typing import TYPE_CHECKING, Awaitable, Mapping, Optional, Protocol, Type, TypeVar
-+from typing import TYPE_CHECKING, Any, Awaitable, Mapping, Optional, Protocol, Type, TypeVar
- 
- import attr
- from aiosignal import Signal
-@@ -64,7 +64,7 @@ class TraceConfig:
-         self._trace_config_ctx_factory = trace_config_ctx_factory
- 
-     def trace_config_ctx(
--        self, trace_request_ctx: Optional[Mapping[str, str]] = None
-+        self, trace_request_ctx: Optional[Mapping[str, Any]] = None
-     ) -> SimpleNamespace:
-         """Return a new trace_config_ctx instance"""
-         return self._trace_config_ctx_factory(trace_request_ctx=trace_request_ctx)
diff --git a/contrib/python/aiohttp/patches/99-rep-get-running-loop.sh b/contrib/python/aiohttp/patches/99-rep-get-running-loop.sh
index b97011e3c67..8565b28f2cf 100644
--- a/contrib/python/aiohttp/patches/99-rep-get-running-loop.sh
+++ b/contrib/python/aiohttp/patches/99-rep-get-running-loop.sh
@@ -1,4 +1,10 @@
 # This patch may be dropped after python 3.13 upver
 
-find aiohttp -type f -exec sed --in-place 's|loop or asyncio.get_running_loop|loop or asyncio.get_event_loop|g' '{}' ';'
+if [ "$(uname)" = "Darwin" ]; then
+    SED=gsed
+else
+    SED=sed
+fi
+
+find aiohttp -type f -exec ${SED} --in-place 's|loop or asyncio.get_running_loop|loop or asyncio.get_event_loop|g' '{}' ';'
 
diff --git a/contrib/python/aiohttp/ya.make b/contrib/python/aiohttp/ya.make
index e714d0fd423..3767e000884 100644
--- a/contrib/python/aiohttp/ya.make
+++ b/contrib/python/aiohttp/ya.make
@@ -2,7 +2,7 @@
 
 PY3_LIBRARY()
 
-VERSION(3.10.6)
+VERSION(3.10.11)
 
 LICENSE(Apache-2.0)
 
@@ -57,7 +57,6 @@ PY_SRCS(
     aiohttp/http_parser.py
     aiohttp/http_websocket.py
     aiohttp/http_writer.py
-    aiohttp/locks.py
     aiohttp/log.py
     aiohttp/multipart.py
     aiohttp/payload.py