Added caching to the async session in request.py and AsyncHTTPProvider

newsfragments/2016.feature.rst

@@ -0,0 +1 @@
+Added session caching to the AsyncHTTPProvider

tests/core/utilities/test_request.py

@@ -1,3 +1,8 @@
+import pytest
+
+from aiohttp import (
+    ClientSession,
+)
 from requests import (
     Session,
     adapters,
@@ -80,3 +85,19 @@ def test_precached_session(mocker):
     assert isinstance(adapter, HTTPAdapter)
     assert adapter._pool_connections == 100
     assert adapter._pool_maxsize == 100
+
+
+@pytest.mark.asyncio
+async def test_async_precached_session(mocker):
+    # Add a session
+    session = ClientSession()
+    request.cache_async_session(URI, session)
+    assert len(request._async_session_cache) == 1
+
+    # Make sure the session isn't duplicated
+    request.cache_async_session(URI, session)
+    assert len(request._async_session_cache) == 1
+
+    # Make sure a request with a different URI addes another cached session
+    request.cache_async_session("{0}/test".format(URI), session)
+    assert len(request._async_session_cache) == 2

web3/_utils/request.py

@@ -27,11 +27,21 @@ def cache_session(endpoint_uri: URI, session: requests.Session) -> None:
     _session_cache[cache_key] = session
 
 
+def cache_async_session(endpoint_uri: URI, session: ClientSession) -> None:
+    cache_key = generate_cache_key(endpoint_uri)
+    _async_session_cache[cache_key] = session
+
+
 def _remove_session(key: str, session: requests.Session) -> None:
     session.close()
 
 
+def _remove_async_session(key: str, session: ClientSession) -> None:
+    session.close()
+
+
 _session_cache = lru.LRU(8, callback=_remove_session)
+_async_session_cache = lru.LRU(8, callback=_remove_async_session)
 
 
 def _get_session(endpoint_uri: URI) -> requests.Session:
@@ -41,6 +51,13 @@ def _get_session(endpoint_uri: URI) -> requests.Session:
     return _session_cache[cache_key]
 
 
+def _get_async_session(endpoint_uri: URI) -> ClientSession:
+    cache_key = generate_cache_key(endpoint_uri)
+    if cache_key not in _async_session_cache:
+        _async_session_cache[cache_key] = ClientSession(raise_for_status=True)
+    return _async_session_cache[cache_key]
+
+
 def make_post_request(endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any) -> bytes:
     kwargs.setdefault('timeout', 10)
     session = _get_session(endpoint_uri)
@@ -55,9 +72,10 @@ async def async_make_post_request(
     endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any
 ) -> bytes:
     kwargs.setdefault('timeout', ClientTimeout(10))
-    async with ClientSession(raise_for_status=True) as session:
-        async with session.post(endpoint_uri,
-                                data=data,
-                                *args,
-                                **kwargs) as response:
-            return await response.read()
+    # https://github.com/ethereum/go-ethereum/issues/17069
+    session = _get_async_session(endpoint_uri)
+    async with session.post(endpoint_uri,
+                            data=data,
+                            *args,
+                            **kwargs) as response:
+        return await response.read()

web3/providers/async_rpc.py

@@ -20,6 +20,7 @@
 )
 from web3._utils.request import (
     async_make_post_request,
+    cache_async_session,
     get_default_http_endpoint,
 )
 from web3.types import (
@@ -39,7 +40,8 @@ class AsyncHTTPProvider(AsyncJSONBaseProvider):
 
     def __init__(
         self, endpoint_uri: Optional[Union[URI, str]] = None,
-            request_kwargs: Optional[Any] = None
+            request_kwargs: Optional[Any] = None,
+            session: Optional[Any] = None
     ) -> None:
         if endpoint_uri is None:
             self.endpoint_uri = get_default_http_endpoint()
@@ -48,6 +50,9 @@ def __init__(
 
         self._request_kwargs = request_kwargs or {}
 
+        if session is not None:
+            cache_async_session(self.endpoint_uri, session)
+
         super().__init__()
 
     def __str__(self) -> str: