Repository: ethereum/web3.py (python)

Issue:
Async concurrency regression with 5.29.0
* Version: 5.29.0
* Python: 3.9.7
* OS: macos

### What was wrong?

Async code with the AsyncHTTPProvider hangs in 5.29.0; previously worked fine in 5.28.0. 

(Shoddy) code sample:
```py
import asyncio
from web3 import Web3, AsyncHTTPProvider
from web3.eth import AsyncEth
from web3.net import AsyncNet
from web3.middleware import async_geth_poa_middleware

networks = {
    'ARBITRUM' : 'https://arb1.arbitrum.io/rpc',
    'AURORA'   : 'https://mainnet.aurora.dev',
    'AVAX'     : 'https://rpc.ankr.com/avalanche/',
    'BSC'      : 'https://bscrpc.com',
    'CELO'     : 'https://rpc.ankr.com/celo/',
    'CRONOS'   : 'https://evm-cronos.crypto.org',
    'FANTOM'   : 'https://rpc.ankr.com/fantom/',
    'FUSE'     : 'https://rpc.fuse.io/',
    'HARMONY'  : 'https://rpc.ankr.com/harmony/',
    'HECO'     : 'https://http-mainnet.hecochain.com',
    'METIS'    : 'https://andromeda.metis.io/?owner=1088',
    'MOONBEAM' : 'https://rpc.api.moonbeam.network',
    'MOONRIVER': 'https://rpc.api.moonriver.moonbeam.network/',
    'POLYGON'  : 'https://polygon-rpc.com/'
}

async def load_networks():
    async def connect_rpc(network, uri):
        rpc = Web3(
            AsyncHTTPProvider(
                uri,
                request_kwargs={'timeout': 10},
            ),
            modules={'eth': AsyncEth, 'net': AsyncNet},
            middlewares=[]
        ) 
        rpc.middleware_onion.inject(async_geth_poa_middleware, layer=0)

        try:
            chain_id = await rpc.eth.chain_id
            print(network, chain_id)
        except Exception as e:
            print(network, e)

    tasks = [connect_rpc(chain, info) for chain, info in networks.items()]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(load_networks())
```

With 5.28.0, the output is as expected.
```
BSC 56
AURORA 1313161554
POLYGON 137
HECO 128
MOONBEAM 1284
MOONRIVER 1285
ARBITRUM 42161
METIS 1088
AVAX 43114
FANTOM 250
CELO 42220
HARMONY 1666600000
FUSE 122
CRONOS 25
```

When updating to 5.29.0, the output hangs. When I abort the program, the error output references a session cache lock (edited for readability)

```
KeyboardInterrupt
Task exception was never retrieved
future: <Task finished name='Task-1' coro=<load_networks() done, defined at []/async_test.py:25> exception=KeyboardInterrupt()>
Traceback (most recent call last):
  File "[]/async_test.py", line 49, in <module>
    asyncio.run(load_networks())
  File "[]/python3.9/asyncio/runners.py", line 47, in run
    _cancel_all_tasks(loop)
  File "[]/python3.9/asyncio/runners.py", line 63, in _cancel_all_tasks
    loop.run_until_complete(
  File "[]/python3.9/asyncio/base_events.py", line 629, in run_until_complete
    self.run_forever()
  File "[]/python3.9/asyncio/base_events.py", line 596, in run_forever
    self._run_once()
  File "[]/python3.9/asyncio/base_events.py", line 1890, in _run_once
    handle._run()
  File "[]/python3.9/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "[]/async_test.py", line 46, in load_networks
    await asyncio.gather(*tasks)
  File "[]/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "[]/python3.9/asyncio/base_events.py", line 629, in run_until_complete
    self.run_forever()
  File "[]/python3.9/asyncio/base_events.py", line 596, in run_forever
    self._run_once()
  File "[]/python3.9/asyncio/base_events.py", line 1890, in _run_once
    handle._run()
  File "[]/python3.9/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "[]/async_test.py", line 38, in connect_rpc
    chain_id = await rpc.eth.chain_id
  File "[]/python3.9/site-packages/web3/eth.py", line 360, in chain_id
    return await self._chain_id()  # type: ignore
  File "[]/python3.9/site-packages/web3/module.py", line 72, in caller
    result = await w3.manager.coro_request(method_str,
  File "[]/python3.9/site-packages/web3/manager.py", line 213, in coro_request
    response = await self._coro_make_request(method, params)
  File "[]/python3.9/site-packages/web3/manager.py", line 160, in _coro_make_request
    return await request_func(method, params)
  File "[]/python3.9/site-packages/web3/middleware/formatting.py", line 137, in middleware
    response = await make_request(method, params)
  File "[]/python3.9/site-packages/web3/providers/async_rpc.py", line 80, in make_request
    raw_response = await async_make_post_request(
  File "[]/python3.9/site-packages/web3/_utils/request.py", line 113, in async_make_post_request
    session = await _get_async_session(endpoint_uri)
  File "[]/python3.9/site-packages/web3/_utils/request.py", line 94, in _get_async_session
    await cache_async_session(endpoint_uri, ClientSession(raise_for_status=True))
  File "[]/python3.9/site-packages/web3/_utils/request.py", line 68, in cache_async_session
    with _async_session_cache_lock:
KeyboardInterrupt
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x106ffafd0>
```

I do see that some SessionCache code [is included in 5.29.0](https://github.com/ethereum/web3.py/compare/v5.28.0...v5.29.0#diff-da1c04440e0604ceb9d314eec7151604f68036eaf4e6206f78f7aa38e36b0a65). My use case may be quite different than what was intended by the Session Cache and is likely related to my report in https://github.com/ethereum/web3.py/pull/2254 (ack, my apologies for not following up on that, @kclowes!). 

Pull request:
[v5] request cache improvements
**Note: Work in progress to skirt what python 3.6 doesn't support**

---

`v5` support related to #2409 and #2690 that is able to support python 3.6:

- Use ``SessionCache`` for sync cache, get rid of LRU cache
- Increase cache size to `100` from `20`
- Use unique cache key identifiers across threads / event loops for async

_Note: A more comprehensive request cache update, especially for async, is the `v6` version. The reason these changes cannot be back-ported to `v5` is that python 3.6 does not support some of the features used in the `v6` async implementation of the cache._

closes #2446 

### Todo:
- [x] Add entry to the [release notes](https://github.com/ethereum/web3.py/blob/master/newsfragments/README.md)

#### Cute Animal Picture

![20221021_094338](https://user-images.githubusercontent.com/3532824/197304634-1f690f11-d6fc-49de-a1d7-642c494c5aa9.jpg)


Patch:
diff --git a/newsfragments/2691.bugfix.rst b/newsfragments/2691.bugfix.rst
new file mode 100644
index 00000000..29b462bd
--- /dev/null
+++ b/newsfragments/2691.bugfix.rst
@@ -0,0 +1,1 @@
+Improve upon issues with session caching and cache locking, especially for multi-threading.
diff --git a/newsfragments/2691.misc.rst b/newsfragments/2691.misc.rst
new file mode 100644
index 00000000..3b99d29c
--- /dev/null
+++ b/newsfragments/2691.misc.rst
@@ -0,0 +1,1 @@
+Increase request session cache size from ``20`` to ``100``.
diff --git a/web3/_utils/request.py b/web3/_utils/request.py
index 8f971a0d..6865f1ed 100644
--- a/web3/_utils/request.py
+++ b/web3/_utils/request.py
@@ -1,11 +1,13 @@
 from collections import (
     OrderedDict,
 )
+import logging
 import os
 import threading
 from typing import (
     Any,
     Dict,
+    List,
 )
 
 from aiohttp import (
@@ -15,16 +17,18 @@ from aiohttp import (
 from eth_typing import (
     URI,
 )
-import lru
 import requests
 
 from web3._utils.caching import (
     generate_cache_key,
 )
 
+logger = logging.getLogger(__name__)
+
+DEFAULT_TIMEOUT = 10
 
-class SessionCache:
 
+class SessionCache:
     def __init__(self, size: int):
         self._size = size
         self._data: OrderedDict[str, Any] = OrderedDict()
@@ -47,6 +51,9 @@ class SessionCache:
     def get_cache_entry(self, key: str) -> Any:
         return self._data[key]
 
+    def clear(self) -> None:
+        self._data.clear()
+
     def __contains__(self, item: str) -> bool:
         return item in self._data
 
@@ -55,64 +62,96 @@ class SessionCache:
 
 
 def get_default_http_endpoint() -> URI:
-    return URI(os.environ.get('WEB3_HTTP_PROVIDER_URI', 'http://localhost:8545'))
+    return URI(os.environ.get("WEB3_HTTP_PROVIDER_URI", "http://localhost:8545"))
+
+
+_session_cache = SessionCache(size=100)
+_session_cache_lock = threading.Lock()
 
 
 def cache_session(endpoint_uri: URI, session: requests.Session) -> None:
-    cache_key = generate_cache_key(endpoint_uri)
-    _session_cache[cache_key] = session
+    # cache key should have a unique thread identifier
+    cache_key = generate_cache_key(f"{threading.get_ident()}:{endpoint_uri}")
+
+    evicted_items = None
+    with _session_cache_lock:
+        if cache_key not in _session_cache:
+            if session is None:
+                session = requests.Session()
+
+            evicted_items = _session_cache.cache(cache_key, session)
+            logger.debug(f"Session cached: {endpoint_uri}, {session}")
+
+    if evicted_items is not None:
+        evicted_sessions = evicted_items.values()
+        for evicted_session in evicted_sessions:
+            logger.debug(
+                f"Session cache full. Session evicted from cache: {evicted_session}",
+            )
+        threading.Timer(
+            DEFAULT_TIMEOUT + 0.1,
+            _close_evicted_sessions,
+            args=[evicted_sessions],
+        ).start()
 
 
-async def cache_async_session(endpoint_uri: URI, session: ClientSession) -> None:
-    cache_key = generate_cache_key(endpoint_uri)
-    with _async_session_cache_lock:
-        evicted_items = _async_session_cache.cache(cache_key, session)
-        if evicted_items is not None:
-            for key, session in evicted_items.items():
-                await session.close()
+def _get_session(endpoint_uri: URI) -> requests.Session:
+    # cache key should have a unique thread identifier
+    cache_key = generate_cache_key(f"{threading.get_ident()}:{endpoint_uri}")
+    if cache_key not in _session_cache:
+        cache_session(endpoint_uri, requests.Session())
+    return _session_cache.get_cache_entry(cache_key)
 
 
-def _remove_session(key: str, session: requests.Session) -> None:
-    session.close()
+def make_post_request(
+    endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any
+) -> bytes:
+    kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
+    session = _get_session(endpoint_uri)
+    # https://github.com/python/mypy/issues/2582
+    response = session.post(endpoint_uri, data=data, *args, **kwargs)  # type: ignore
+    response.raise_for_status()
+
+    return response.content
 
 
-_session_cache = lru.LRU(8, callback=_remove_session)
+def _close_evicted_sessions(evicted_sessions: List[requests.Session]) -> None:
+    for evicted_session in evicted_sessions:
+        evicted_session.close()
+        logger.debug(f"Closed evicted session: {evicted_session}")
+
+
+# --- async --- #
+
+_async_session_cache = SessionCache(size=100)
 _async_session_cache_lock = threading.Lock()
-_async_session_cache = SessionCache(size=20)
 
 
-def _get_session(endpoint_uri: URI) -> requests.Session:
-    cache_key = generate_cache_key(endpoint_uri)
-    if cache_key not in _session_cache:
-        _session_cache[cache_key] = requests.Session()
-    return _session_cache[cache_key]
+async def cache_async_session(endpoint_uri: URI, session: ClientSession) -> None:
+    # cache key should have a unique thread identifier
+    cache_key = generate_cache_key(f"{threading.get_ident()}:{endpoint_uri}")
+
+    with _async_session_cache_lock:
+        evicted_items = _async_session_cache.cache(cache_key, session)
+        if evicted_items is not None:
+            for key, session in evicted_items.items():
+                await session.close()
 
 
 async def _get_async_session(endpoint_uri: URI) -> ClientSession:
-    cache_key = generate_cache_key(endpoint_uri)
+    # cache key should have a unique thread identifier
+    cache_key = generate_cache_key(f"{threading.get_ident()}:{endpoint_uri}")
+
     if cache_key not in _async_session_cache:
         await cache_async_session(endpoint_uri, ClientSession(raise_for_status=True))
     return _async_session_cache.get_cache_entry(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)
-    # https://github.com/python/mypy/issues/2582
-    response = session.post(endpoint_uri, data=data, *args, **kwargs)  # type: ignore
-    response.raise_for_status()
-
-    return response.content
-
-
 async def async_make_post_request(
     endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any
 ) -> bytes:
-    kwargs.setdefault('timeout', ClientTimeout(10))
+    kwargs.setdefault("timeout", ClientTimeout(DEFAULT_TIMEOUT))
     # https://github.com/ethereum/go-ethereum/issues/17069
     session = await _get_async_session(endpoint_uri)
-    async with session.post(endpoint_uri,
-                            data=data,
-                            *args,
-                            **kwargs) as response:
+    async with session.post(endpoint_uri, data=data, *args, **kwargs) as response:
         return await response.read()


Test patch:
diff --git a/tests/core/utilities/test_request.py b/tests/core/utilities/test_request.py
index 32d4cf9b..e55bd7ba 100644
--- a/tests/core/utilities/test_request.py
+++ b/tests/core/utilities/test_request.py
@@ -1,8 +1,17 @@
+import asyncio
+from concurrent.futures import (
+    ThreadPoolExecutor,
+)
 import pytest
+import threading
+import time
 
 from aiohttp import (
     ClientSession,
 )
+from eth_typing import (
+    URI,
+)
 from requests import (
     Session,
     adapters,
@@ -15,6 +24,9 @@ from requests.adapters import (
 from web3._utils import (
     request,
 )
+from web3._utils.caching import (
+    generate_cache_key,
+)
 from web3._utils.request import (
     SessionCache,
 )
@@ -22,8 +34,8 @@ from web3._utils.request import (
 
 class MockedResponse:
     def __init__(self, text="", status_code=200):
-        assert (isinstance(text, str))
-        assert (isinstance(status_code, int))
+        assert isinstance(text, str)
+        assert isinstance(status_code, int)
         self.status_code = status_code
         self.ok = 200 <= status_code < 400
         self.text = text
@@ -34,7 +46,14 @@ class MockedResponse:
         pass
 
 
-URI = "http://mynode.local:8545"
+TEST_URI = URI("http://mynode.local:8545")
+UNIQUE_URIS = [
+    "https://www.test1.com",
+    "https://www.test2.com",
+    "https://www.test3.com",
+    "https://www.test4.com",
+    "https://www.test5.com",
+]
 
 
 def check_adapters_mounted(session: Session):
@@ -42,25 +61,37 @@ def check_adapters_mounted(session: Session):
     assert len(session.adapters) == 2
 
 
+def _simulate_call(uri):
+    _session = request._get_session(uri)
+
+    # simulate a call taking 0.01s to return a response
+    time.sleep(0.01)
+    return _session
+
+
 def test_make_post_request_no_args(mocker):
     mocker.patch("requests.Session.post", return_value=MockedResponse())
     request._session_cache.clear()
 
     # Submit a first request to create a session with default parameters
     assert len(request._session_cache) == 0
-    response = request.make_post_request(URI, b'request')
+    response = request.make_post_request(TEST_URI, data=b"request")
     assert response == "content"
     assert len(request._session_cache) == 1
-    session = request._session_cache.values()[0]
-    session.post.assert_called_once_with(URI, data=b'request', timeout=10)
+    cache_key = generate_cache_key(f"{threading.get_ident()}:{TEST_URI}")
+    session = request._session_cache.get_cache_entry(cache_key)
+    session.post.assert_called_once_with(TEST_URI, data=b"request", timeout=10)
 
     # Ensure the adapter was created with default values
     check_adapters_mounted(session)
-    adapter = session.get_adapter(URI)
+    adapter = session.get_adapter(TEST_URI)
     assert isinstance(adapter, HTTPAdapter)
     assert adapter._pool_connections == DEFAULT_POOLSIZE
     assert adapter._pool_maxsize == DEFAULT_POOLSIZE
 
+    # clear cache
+    request._session_cache.clear()
+
 
 def test_precached_session(mocker):
     mocker.patch("requests.Session.post", return_value=MockedResponse())
@@ -68,46 +99,32 @@ def test_precached_session(mocker):
     # Update the cache with a handcrafted session
     adapter = adapters.HTTPAdapter(pool_connections=100, pool_maxsize=100)
     session = Session()
-    session.mount('http://', adapter)
-    session.mount('https://', adapter)
-    request.cache_session(URI, session)
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+    request.cache_session(TEST_URI, session)
 
     # Submit a second request with different arguments
     assert len(request._session_cache) == 1
-    response = request.make_post_request(URI, b'request', timeout=60)
+    response = request.make_post_request(TEST_URI, data=b"request", timeout=60)
     assert response == "content"
     assert len(request._session_cache) == 1
 
     # Ensure the timeout was passed to the request
-    session = request._get_session(URI)
-    session.post.assert_called_once_with(URI, data=b'request', timeout=60)
+    session = request._get_session(TEST_URI)
+    session.post.assert_called_once_with(TEST_URI, data=b"request", timeout=60)
 
     # Ensure the adapter parameters match those we specified
     check_adapters_mounted(session)
-    adapter = session.get_adapter(URI)
+    adapter = session.get_adapter(TEST_URI)
     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()
-    await request.cache_async_session(URI, session)
-    assert len(request._async_session_cache) == 1
-
-    # Make sure the session isn't duplicated
-    await request.cache_async_session(URI, session)
-    assert len(request._async_session_cache) == 1
-
-    # Make sure a request with a different URI adds another cached session
-    await request.cache_async_session(f"{URI}/test", session)
-    assert len(request._async_session_cache) == 2
+    # clear cache
+    request._session_cache.clear()
 
 
 def test_cache_session_class():
-
     cache = SessionCache(2)
     evicted_items = cache.cache("1", "Hello1")
     assert cache.get_cache_entry("1") == "Hello1"
@@ -126,9 +143,132 @@ def test_cache_session_class():
     evicted_items = cache.cache("3", "Hello3")
     assert "2" in cache
     assert "3" in cache
+
     assert "1" not in cache
+    assert "1" in evicted_items
 
     with pytest.raises(KeyError):
         # This should throw a KeyError since the cache size was 2 and 3 were inserted
         # the first inserted cached item was removed and returned in evicted items
         cache.get_cache_entry("1")
+
+    # clear cache
+    request._session_cache.clear()
+
+
+def test_cache_does_not_close_session_before_a_call_when_multithreading():
+    # save default values
+    session_cache_default = request._session_cache
+    timeout_default = request.DEFAULT_TIMEOUT
+
+    # set cache size to 1 + set future session close thread time to 0.01s
+    request._session_cache = SessionCache(1)
+    _timeout_for_testing = 0.01
+    request.DEFAULT_TIMEOUT = _timeout_for_testing
+
+    with ThreadPoolExecutor(max_workers=len(UNIQUE_URIS)) as exc:
+        all_sessions = [exc.submit(_simulate_call, uri) for uri in UNIQUE_URIS]
+
+    # assert last session remains in cache, all others evicted
+    cache_data = request._session_cache._data
+    assert len(cache_data) == 1
+    _key, cached_session = cache_data.popitem()
+    assert cached_session == all_sessions[-1].result()  # result of the `Future`
+
+    # -- teardown -- #
+
+    # close the cached session before exiting test
+    cached_session.close()
+
+    # reset default values
+    request._session_cache = session_cache_default
+    request.DEFAULT_TIMEOUT = timeout_default
+
+    # clear cache
+    request._session_cache.clear()
+
+
+def test_unique_cache_keys_created_per_thread_with_same_uri():
+    # somewhat inspired by issue #2680
+
+    with ThreadPoolExecutor(max_workers=2) as exc:
+        test_sessions = [exc.submit(_simulate_call, TEST_URI) for _ in range(2)]
+
+    # assert unique keys are generated per thread for the same uri
+    assert len(request._session_cache._data) == 2
+
+    # -- teardown -- #
+
+    # appropriately close the test sessions
+    [session.result().close() for session in test_sessions]
+
+    # clear cache
+    request._session_cache.clear()
+
+
+# -- async -- #
+
+
+@pytest.mark.asyncio
+async def test_async_precached_session():
+    # Add a session
+    session = ClientSession()
+    await request.cache_async_session(TEST_URI, session)
+    assert len(request._async_session_cache) == 1
+
+    # Make sure the session isn't duplicated
+    await request.cache_async_session(TEST_URI, session)
+    assert len(request._async_session_cache) == 1
+
+    # Make sure a request with a different URI adds another cached session
+    await request.cache_async_session(URI(f"{TEST_URI}/test"), session)
+    assert len(request._async_session_cache) == 2
+
+    # -- teardown -- #
+
+    # appropriately close the cached sessions
+    [await session.close() for session in request._async_session_cache._data.values()]
+
+    # clear cache
+    request._async_session_cache.clear()
+
+
+@pytest.mark.asyncio
+async def test_async_unique_cache_keys_created_per_thread_with_same_uri():
+    # inspired by issue #2680
+    # Note: unique event loops for each thread are important here
+
+    # capture the test sessions to appropriately close them in teardown
+    test_sessions = []
+
+    def target_function(endpoint_uri):
+        event_loop = asyncio.new_event_loop()
+        unique_session = event_loop.run_until_complete(
+            request._get_async_session(endpoint_uri)
+        )
+        test_sessions.append(unique_session)
+
+    threads = []
+
+    for _ in range(2):
+        thread = threading.Thread(
+            target=target_function,
+            args=(TEST_URI,),
+            daemon=True,
+        )
+        thread.start()
+        threads.append(thread)
+
+    [thread.join() for thread in threads]
+
+    # assert unique keys are generated per thread, w/ unique event loops,
+    # for the same uri
+    assert len(request._async_session_cache._data) == 2
+
+    # -- teardown -- #
+
+    # appropriately close the test sessions
+    [await session.close() for session in test_sessions]
+
+    # clear cache
+    request._async_session_cache.clear()