Repository: ethereum/web3.py

Issue:
script coredumps when using threads 
* Version: 5.28.0 (tried to downgrade to version 5.9, same result )
* Python: 3.10.3 (also reproduces on earlier versions) 
* OS: linux
* `pip freeze` output

```
acme==1.22.0
aiohttp==3.8.1
aiosignal==1.2.0
ansi2html==1.6.0
ansible==2.9.27
appdirs==1.4.4
argcomplete==1.12.3
asgiref==3.4.1
asn1crypto==1.4.0
async-generator==1.10
async-timeout==4.0.2
attrs==21.2.0
Automat==20.2.0
Babel==2.9.1
backcall==0.1.0
base58==2.1.1
bcrypt==3.2.0
Beaker==1.10.0
beautifulsoup4==4.9.3
bitarray==1.2.2
blivet==3.4.3
blivet-gui==2.3.0
Bottleneck==1.3.2
Brlapi==0.8.2
Brotli==1.0.9
cached-property==1.5.2
cairocffi==1.3.0
CairoSVG==2.5.2
certbot==1.22.0
certifi==2020.12.5
cffi==1.15.0
chardet==4.0.0
charset-normalizer==2.0.4
chrome-gnome-shell==0.0.0
click==8.0.1
coincurve==16.0.0
colorama==0.4.4
ConfigArgParse==1.4.1
configobj==5.0.6
confluent-kafka==0.11.6
constantly==15.1.0
cryptography==3.4.7
cssselect2==0.3.0
cupshelpers==1.0
cycler==0.10.0
cytoolz==0.11.2
dasbus==1.6
dateparser==1.1.0
dbus-python==1.2.18
decorator==5.0.9
defusedxml==0.7.1
distro==1.6.0
Django==3.2.12
dnspython==2.1.0
docker==5.0.2
docker-compose==1.29.2
docker-pycreds==0.4.0
docker-squash==1.0.8
dockerpty==0.4.1
docopt==0.6.2
eth-abi==2.1.1
eth-account==0.5.7
eth-hash==0.3.2
eth-keyfile==0.5.1
eth-keys==0.3.4
eth-rlp==0.2.1
eth-typing==2.2.2
eth-utils==1.10.0
ethereum==2.3.2
execnet==1.9.0
fastavro==1.4.9
fedora-third-party==0.8
file-magic==0.4.0
flake8==4.0.1
fonttools==4.26.1
fros==1.1
frozenlist==1.2.0
fs==2.4.11
future==0.18.2
gpg==1.15.1
grafanalib==0.6.2
greenlet==1.1.1
h11==0.12.0
hexbytes==0.2.2
httplib2==0.20.2
humanize==0.5.1
hyperlink==21.0.0
idna==3.2
incremental==21.3.0
iniconfig==1.1.1
initial-setup==0.3.93
ipfshttpclient==0.8.0a2
ipython==7.26.0
ipython_genutils==0.1.0
isc==2.0
jedi==0.18.0
Jinja2==3.0.1
jira==2.0.0
jmespath==0.10.0
josepy==1.9.0
jsonrpcclient==4.0.2
jsonschema==3.2.0
kazoo==2.8.0
kiwisolver==1.4.0
langtable==0.0.57
libcomps==0.1.18
libvirt-python==7.6.0
lru-dict==1.1.7
lxml==4.6.5
Mako==1.1.4
Markdown==3.3.6
MarkupSafe==2.0.0
matplotlib==3.5.1
matplotlib-inline==0.1.2
mccabe==0.6.1
mercurial==5.9.3
multiaddr==0.0.9
multidict==5.2.0
munkres==1.1.2
mypy-extensions==0.4.3
netaddr==0.8.0
nftables==0.1
ntlm-auth==1.5.0
numexpr==2.7.1
numpy==1.21.5
oauthlib==3.0.2
olefile==0.46
outcome==1.1.0
packaging==21.3
pandas==1.3.3
pandas-datareader==0.10.0
paramiko==2.9.1
parsedatetime==2.6
parsimonious==0.8.1
parso==0.8.2
Paste==3.5.0
pbkdf2==1.3
pbr==5.6.0
pexpect==4.8.0
pickleshare==0.7.5
pid==2.2.3
Pillow==8.3.2
pluggy==0.13.1
ply==3.11
productmd==1.33
prometheus-client==0.12.0
prompt-toolkit==3.0.24
protobuf==3.19.1
ptyprocess==0.6.0
pwquality==1.4.4
py==1.11.0
py-ecc==5.2.0
py3nvml==0.2.6
pyasn1==0.4.8
pyasn1-modules==0.2.8
PyAudio==0.2.11
pycairo==1.20.1
pycodestyle==2.8.0
pycparser==2.20
pycrypto==2.6.1
pycryptodome==3.12.0
pycups==2.0.1
pycurl==7.44.1
pyenchant==3.2.2
pyethash==0.1.27
pyflakes==2.4.0
pygame==2.1.2
Pygments==2.9.0
PyGObject==3.42.0
PyJWT==2.1.0
pykickstart==3.34
PyNaCl==1.4.0
pyOpenSSL==21.0.0
pyparsing==2.4.7
pyparted==3.12.0
PyQt5==5.15.6
PyQt5-sip==12.9.0
PyQtWebEngine==5.15.2
pyRFC3339==1.1
pyrsistent==0.18.1
pysha3==1.0.2
PySocks==1.7.1
pystray==0.17.3
pytest==6.2.4
pytest-forked==1.3.0
pytest-xdist==2.3.0
python-augeas==1.1.0
python-binance==1.0.15
python-dateutil==2.8.1
python-dotenv==0.18.0
python-gettext==4.0
python-manatools==0.0.3
python-meh==0.50
python-memcached==1.59
python-xlib==0.31
pytz==2021.3
pytz-deprecation-shim==0.1.0.post0
pyudev==0.22.0
pywinrm==0.4.1
pyxdg==0.27
PyYAML==5.4.1
rancher-gitlab-deploy==1.7
regex==2022.3.2
repoze.lru==0.7
requests==2.27.0
requests-file==1.5.1
requests-ftp==0.3.1
requests-ntlm==1.1.0
requests-oauthlib==1.3.0
requests-toolbelt==0.9.1
rlp==1.2.0
rpm==4.17.0
rpmconf==1.1.4
schedule==1.1.0
scipy==1.7.3
scrypt==0.8.19
selenium==4.1.3
selinux==3.3
Send2Trash==1.7.1
sepolicy==3.3
service-identity==21.1.0
setools==4.4.0
setproctitle==1.2.2
shtab==1.5.2
simpleaudio==1.0.4
simplejson==3.17.2
simpleline==1.8.2
six==1.16.0
sniffio==1.2.0
sortedcontainers==2.4.0
sos==4.2
soupsieve==2.3.1
SQLAlchemy==1.4.32
sqlparse==0.4.1
systemd-coredump-python==2
systemd-python==234
tables==3.6.1
Tempita==0.5.2
termcolor==1.1.0
texttable==1.6.2
tinycss2==1.1.1
tldr==3.1.0
toml==0.10.2
toolz==0.11.2
traitlets==5.0.5
trio==0.19.0
trio-websocket==0.9.2
Twisted==21.7.0
typing-extensions==3.10.0.2
tzdata==2021.5
tzlocal==4.1
ujson==5.1.0
urllib3==1.26.7
varint==1.0.2
wcwidth==0.2.5
web3==5.28.0
webencodings==0.5.1
websocket-client==1.2.3
websockets==9.1
wsproto==1.0.0
xcffib==0.11.1
xlrd==2.0.1
xlwt==1.3.0
xmltodict==0.12.0
yarl==1.7.2
zope.component==4.3.0
zope.event==4.2.0
zope.interface==5.4.0
```


With this set of RPCs, the script falls into a coredump on cycles 3-10. 
```
andreyc@andrnote ~/devel $ cat thread5c.py 
import time
from threading import Thread
from web3 import Web3


def run(url, account):
    w3 = Web3(Web3.HTTPProvider(url))
    balance = w3.eth.getBalance(account)
    print("account {}, balance {}".format(account, balance))


addr = "0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d"
while True:
    for url, account in (("https://rpc-mainnet.lachain.io/", addr),
                         ("https://mainnet.infura.io/v3/9aa3d95b3bc440fa88ea12eaa4456161", addr),
                         ("https://polygon-rpc.com/", addr),
                         ("https://bsc-dataseed.binance.org/", addr),
                         ("https://api.avax.network/ext/bc/C/rpc", addr),
                         ("https://rpc-testnet.lachain.io/", addr),
                         ("https://ropsten.infura.io/v3/9a6feb0c6140449c9997d5f98b2a3665", addr),
                         ("https://matic-mumbai.chainstacklabs.com/", addr),
                         ("https://data-seed-prebsc-1-s2.binance.org:8545/", addr),
                         ("https://api.avax-test.network/ext/bc/C/rpc", addr)):
         Thread(target=run, args=(url, account)).start()
    time.sleep(0.1)
andreyc@andrnote ~/devel $ python thread5c.py 
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 2019591593909653812
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 2019591593909653812
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 1000000000000000000
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 1000000000000000000
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
Exception in thread Thread-54 (run):
account 0xF474Cf03ccEfF28aBc65C9cbaE594F725c80e12d, balance 0
Segmentation fault (core dumped)
```

gdb-py-bt:
```
(gdb) py-bt
Traceback (most recent call first):
  File "/home/andreyc/.local/lib/python3.10/site-packages/eth_utils/address.py", line 118, in <genexpr>
    if int(address_hash[i], 16) > 7
  File "/home/andreyc/.local/lib/python3.10/site-packages/eth_utils/address.py", line 115, in to_checksum_address
    "".join(
  File "/home/andreyc/.local/lib/python3.10/site-packages/eth_utils/address.py", line 134, in is_checksum_address
    is_equal = value == to_checksum_address(value)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/validation.py", line 180, in validate_address
    if not is_checksum_address(value):
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/normalizers.py", line 196, in abi_address_to_hex
    validate_address(data)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/normalizers.py", line 78, in wrapper
    modified = to_wrap(type_str, data)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/abi.py", line 842, in map_to_typed_data
    return ABITypedData(func(*elements))
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/formatters.py", line 90, in recursive_map
    return func(items_mapped)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/decorators.py", line 30, in wrapped
    wrapped_val = to_wrap(*args)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/formatters.py", line 88, in recurse
    return recursive_map(func, item)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/formatters.py", line 76, in map_collection
    return datatype(map(func, collection))
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/formatters.py", line 89, in recursive_map
    items_mapped = map_collection(recurse, data)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/decorators.py", line 30, in wrapped
    wrapped_val = to_wrap(*args)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/abi.py", line 845, in data_tree_map
    return recursive_map(map_to_typed_data, data_tree)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/_utils/abi.py", line 811, in map_abi_data
    return pipe(data, *pipeline)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/formatting.py", line 93, in middleware
    params = formatter(params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/formatting.py", line 94, in middleware
    response = make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/formatting.py", line 94, in middleware
    response = make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/attrdict.py", line 33, in middleware
    response = make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/formatting.py", line 94, in middleware
    response = make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/gas_price_strategy.py", line 90, in middleware
    return make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/middleware/formatting.py", line 94, in middleware
    response = make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/manager.py", line 150, in _make_request
    return request_func(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/manager.py", line 197, in request_blocking
    response = self._make_request(method, params)
  File "/home/andreyc/.local/lib/python3.10/site-packages/web3/module.py", line 57, in caller
    result = w3.manager.request_blocking(method_str,
  File "/home/andreyc/devel/thread5c.py", line 8, in run
    balance = w3.eth.getBalance(account)
  File "/usr/lib64/python3.10/threading.py", line 946, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib64/python3.10/threading.py", line 1009, in _bootstrap_inner
    self.run()
  File "/usr/lib64/python3.10/threading.py", line 966, in _bootstrap
    self._bootstrap_inner()
```


Discussion:
kclowes: Thanks @dbfreem! We're all neck deep in other things right now, but will review soon :tm:
dbfreem: Hey @fselmo this looks good to go. This should get the blocking request in good shape in regards to threading but there is still an issue with the async request and threading. This is documented in issue #2446. At least, this refactor of request will make the fix for #2446 easier once a direction is decided. 

Do you want me to fix that merge conflict? I would go ahead and fix it but wasn't sure where you stood on the rebase. 
fselmo: > This should get the blocking request in good shape in regards to threading but there is still an issue with the async request and threading. This is documented in issue https://github.com/ethereum/web3.py/issues/2446. At least, this refactor of request.py will make the fix for https://github.com/ethereum/web3.py/issues/2446 easier once a direction is decided.

Sounds good. I just wanted to get this cleaned up and rebased so we can keep the conversation going. I am going to spend a bit more time across these PRs but just wanted to make sure it was rebased appropriately. Thanks 👌 

> Do you want me to fix that merge conflict? I would go ahead and fix it but wasn't sure where you stood on the rebase, let me know which way you want to go. 

I took care of the merge conflict, thanks. We've been merging a lot of formatting changes this week.
fselmo: @dbfreem I added a commit to this PR that attempts to resolve some of what we have been discussing. Particularly, the race condition  issue with many sessions running at once. If the cache filled too quickly then the first session, let's say, would be evicted before it ever has time to make a call. Issuing a task to run at `DEFAULT_TIMEOUT + (some arbitrarily small amount of time)` allows for ample time to make the call before that session is closed.

I haven't tinkered this much with async so I'm hoping we can chat about it here. I added a test that was failing up until before this commit that basically simulates the conditions I was seeing from #2446. This commit seems to resolve that even for small cache size (even for `size=1` which is a pretty extreme case).

I ran with a similar approach as to one you'd posted in making the lock actually async by using the `@asynccontextmanager` and a `ThreadPoolExecutor`. Thoughts on this approach?
dbfreem: This is an interesting approach. I started reading [this](https://developpaper.com/what-are-the-common-problems-with-caching-in-high-concurrency-scenarios/) to see if anyone else has run into these types of issue in high concurrency situations. It made me think that fundamentally, if I get a session from the cache I want to make sure that session is not modified until I am done with it. The reason I bring this up is that even with the approaches already implemented in this PR my session could still get closed while I am using it on the sync and async side. 

For instance image I call the [get_session](https://github.com/ethereum/web3.py/blob/806b513d3ad3e8602ab6832aa34b9ba4dff5d8d1/web3/_utils/request.py#L85) method. I should be assured that the session I get back from it will stay open until I am done with it. Right now that guarantee doesn't exist. It is better with this PR but still not perfect. I actually think after reading that article and and thinking through this that the lock should be moved to the [get_response_from_post_request](https://github.com/ethereum/web3.py/blob/806b513d3ad3e8602ab6832aa34b9ba4dff5d8d1/web3/_utils/request.py#L111) method. This would assure that the thread is done with the session before something else could close it. 

So that is for sync but what about async? For Async I think there is a fundamental question of do we want to support async multithreaded code. Would someone be running async in multiple threads if not the the async lock may be appropriate shown [here](https://docs.python.org/3/library/asyncio-sync.html#asyncio.Lock). Is there some forum where this question could be asked to consumers of the library? If we will have someone running async in multiple threads then I can't currently think of another way to make this work then what you have proposed. I still need to read through your commit in a little more detail.

I know that's a lot of info, thoughts??
fselmo: Yeah I think I'd be good with moving the lock to the whole request for sync. I was headed that direction with both but wanted to see if we could tackle multithreading for async first. I think this commit along with allowing the cache size to be customized might be a really good combination answer to our problems. If someone expects many threads for their application and they increase the cache size to match it, they should never run into any issues... and if they still exceed the cache size, this commit should take care of that situation. Thoughts on that?

> Would someone be running async in multiple threads if not the the async lock may be appropriate shown [here](https://docs.python.org/3/library/asyncio-sync.html#asyncio.Lock).

Curious what the difference is between `asyncio.Lock` and the one implemented in my commit ([source](https://stackoverflow.com/questions/63420413/how-to-use-threading-lock-in-async-function-while-object-can-be-accessed-from-mu)).

> The reason I bring this up is that even with the approaches already implemented in this PR my session could still get closed while I am using it on the sync and async side.

Yeah I think sync can be addressed by moving the lock to the request, as we mentioned above. 

For async, can we try to follow a scenario where the async session would be closed before you're done with it? The way I'm seeing this change, and maybe we can even extend the time on the timer thread to make sure of this, is that if a session is evicted from the cache, it is held for at least the amount of time it takes to finish the original call (`DEFAULT_TIMEOUT`). If a new call is made to the same url and it isn't in the cache, it would create a new session in the cache for that URL. Am I missing something there?
dbfreem: > For async, can we try to follow a scenario where the async session would be closed before you're done with it? The way I'm seeing this change, and maybe we can even extend the time on the timer thread to make sure of this, is that if a session is evicted from the cache, it is held for at least the amount of time it takes to finish the original call (DEFAULT_TIMEOUT). If a new call is made to the same url and it isn't in the cache, it would create a new session in the cache for that URL. Am I missing something there?

@fselmo after reading through what you have on the async side I "think" it might work. It is hard to say without running the code, and I haven't had much time lately. Sorry. 
fselmo: > @fselmo after reading through what you have on the async side I "think" it might work. It is hard to say without running the code, and I haven't had much time lately. Sorry. 

Totally get it, no worries at all. Was just curious to pick your brain about it and run through all the possible scenarios. We can probably add a bit more to the timeout too just to make sure. I'll push up the sync changes we talked about too and try to think of some more tests. Then we can do one last pass through... but I think this is a big improvement on what we have. Thanks again for getting it going.

fselmo: @kclowes, I think this is ready for a full review now. Maybe checking the latest commits by themselves so you can see all that changed since the last review.

Patch:
diff --git a/newsfragments/2409.bugfix.rst b/newsfragments/2409.bugfix.rst
new file mode 100644
index 00000000..fe997cd6
--- /dev/null
+++ b/newsfragments/2409.bugfix.rst
@@ -0,0 +1,1 @@
+Improve upon issues with session caching - better support for multithreading and make sure session eviction from cache does not happen prematurely.
\ No newline at end of file
diff --git a/web3/_utils/request.py b/web3/_utils/request.py
index 242e20cb..5a3726af 100644
--- a/web3/_utils/request.py
+++ b/web3/_utils/request.py
@@ -1,11 +1,20 @@
+import asyncio
 from collections import (
     OrderedDict,
 )
+from concurrent.futures import (
+    ThreadPoolExecutor,
+)
+import contextlib
+import logging
 import os
 import threading
 from typing import (
     Any,
+    AsyncGenerator,
     Dict,
+    List,
+    Optional,
     Union,
 )
 
@@ -17,13 +26,16 @@ 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:
     def __init__(self, size: int):
@@ -48,6 +60,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
 
@@ -59,30 +74,46 @@ def get_default_http_endpoint() -> URI:
     return URI(os.environ.get("WEB3_HTTP_PROVIDER_URI", "http://localhost:8545"))
 
 
-def cache_session(endpoint_uri: URI, session: requests.Session) -> None:
-    cache_key = generate_cache_key(endpoint_uri)
-    _session_cache[cache_key] = session
+_session_cache = SessionCache(size=20)
+_session_cache_lock = threading.Lock()
 
 
-def _remove_session(_key: str, session: requests.Session) -> None:
-    session.close()
+def cache_and_return_session(
+    endpoint_uri: URI, session: requests.Session = None
+) -> requests.Session:
+    cache_key = generate_cache_key(endpoint_uri)
 
+    evicted_items = None
+    with _session_cache_lock:
+        if cache_key not in _session_cache:
+            if session is None:
+                session = requests.Session()
 
-_session_cache = lru.LRU(8, callback=_remove_session)
+            evicted_items = _session_cache.cache(cache_key, session)
+            logger.debug(f"Session cached: {endpoint_uri}, {session}")
 
+        cached_session = _session_cache.get_cache_entry(cache_key)
 
-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]
+    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()
+
+    return cached_session
 
 
 def get_response_from_get_request(
     endpoint_uri: URI, *args: Any, **kwargs: Any
 ) -> requests.Response:
-    kwargs.setdefault("timeout", 10)
-    session = get_session(endpoint_uri)
+    kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
+    session = cache_and_return_session(endpoint_uri)
     response = session.get(endpoint_uri, *args, **kwargs)
     return response
 
@@ -90,8 +121,8 @@ def get_response_from_get_request(
 def get_response_from_post_request(
     endpoint_uri: URI, *args: Any, **kwargs: Any
 ) -> requests.Response:
-    kwargs.setdefault("timeout", 10)
-    session = get_session(endpoint_uri)
+    kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
+    session = cache_and_return_session(endpoint_uri)
     response = session.post(endpoint_uri, *args, **kwargs)
     return response
 
@@ -104,33 +135,77 @@ def make_post_request(
     return response.content
 
 
+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_lock = threading.Lock()
+
 _async_session_cache = SessionCache(size=20)
+_async_session_cache_lock = threading.Lock()
+_pool = ThreadPoolExecutor(max_workers=1)
 
 
-async def cache_async_session(endpoint_uri: URI, session: ClientSession) -> None:
+async def cache_and_return_async_session(
+    endpoint_uri: URI,
+    session: Optional[ClientSession] = None,
+) -> ClientSession:
     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()
 
-
-async def get_async_session(endpoint_uri: URI) -> ClientSession:
-    cache_key = generate_cache_key(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)
+    evicted_items = None
+    async with async_lock(_async_session_cache_lock):
+        if cache_key not in _async_session_cache:
+            if session is None:
+                session = ClientSession(raise_for_status=True)
+
+            evicted_items = _async_session_cache.cache(cache_key, session)
+            logger.debug(f"Async session cached: {endpoint_uri}, {session}")
+
+        cached_session = _async_session_cache.get_cache_entry(cache_key)
+
+    if evicted_items is not None:
+        # At this point the evicted sessions are already popped out of the cache and
+        # just stored in the `evicted_sessions` dict. So we can kick off a future task
+        # to close them and it should be safe to pop out of the lock here.
+        evicted_sessions = evicted_items.values()
+        for evicted_session in evicted_sessions:
+            logger.debug(
+                "Async session cache full. Session evicted from cache: "
+                f"{evicted_session}",
+            )
+        # Kick off a future task, in a separate thread, to close the evicted
+        # sessions. In the case that the cache filled very quickly and some
+        # sessions have been evicted before their original request has been made,
+        # we set the timer to a bit more than the `DEFAULT_TIMEOUT` for a call. This
+        # should make it so that any call from an evicted session can still be made
+        # before the session is closed.
+        threading.Timer(
+            DEFAULT_TIMEOUT + 0.1,
+            _async_close_evicted_sessions,
+            args=[evicted_sessions],
+        ).start()
+
+    return cached_session
+
+
+@contextlib.asynccontextmanager
+async def async_lock(lock: threading.Lock) -> AsyncGenerator[None, None]:
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(_pool, lock.acquire)
+    try:
+        yield
+    finally:
+        lock.release()
 
 
 async def async_get_response_from_get_request(
     endpoint_uri: URI, *args: Any, **kwargs: Any
 ) -> ClientResponse:
-    kwargs.setdefault("timeout", ClientTimeout(10))
-    session = await get_async_session(endpoint_uri)
+    kwargs.setdefault("timeout", ClientTimeout(DEFAULT_TIMEOUT))
+    session = await cache_and_return_async_session(endpoint_uri)
     response = await session.get(endpoint_uri, *args, **kwargs)
     return response
 
@@ -138,9 +213,10 @@ async def async_get_response_from_get_request(
 async def async_get_response_from_post_request(
     endpoint_uri: URI, *args: Any, **kwargs: Any
 ) -> ClientResponse:
-    kwargs.setdefault("timeout", ClientTimeout(10))
-    session = await get_async_session(endpoint_uri)
+    kwargs.setdefault("timeout", ClientTimeout(DEFAULT_TIMEOUT))
+    session = await cache_and_return_async_session(endpoint_uri)
     response = await session.post(endpoint_uri, *args, **kwargs)
+
     return response
 
 
@@ -157,3 +233,17 @@ async def async_get_json_from_client_response(
     response: ClientResponse,
 ) -> Dict[str, Any]:
     return await response.json()
+
+
+def _async_close_evicted_sessions(evicted_sessions: List[ClientSession]) -> None:
+    loop = asyncio.new_event_loop()
+
+    for evicted_session in evicted_sessions:
+        loop.run_until_complete(evicted_session.close())
+        logger.debug(f"Closed evicted async session: {evicted_session}")
+
+    if any(not evicted_session.closed for evicted_session in evicted_sessions):
+        logger.warning(
+            f"Some evicted async sessions were not properly closed: {evicted_sessions}"
+        )
+    loop.close()
diff --git a/web3/providers/async_rpc.py b/web3/providers/async_rpc.py
index aa92f046..258f585b 100644
--- a/web3/providers/async_rpc.py
+++ b/web3/providers/async_rpc.py
@@ -23,7 +23,7 @@ from web3._utils.http import (
 )
 from web3._utils.request import (
     async_make_post_request,
-    cache_async_session as _cache_async_session,
+    cache_and_return_async_session as _cache_and_return_async_session,
     get_default_http_endpoint,
 )
 from web3.types import (
@@ -55,8 +55,8 @@ class AsyncHTTPProvider(AsyncJSONBaseProvider):
 
         super().__init__()
 
-    async def cache_async_session(self, session: ClientSession) -> None:
-        await _cache_async_session(self.endpoint_uri, session)
+    async def cache_async_session(self, session: ClientSession) -> ClientSession:
+        return await _cache_and_return_async_session(self.endpoint_uri, session)
 
     def __str__(self) -> str:
         return f"RPC connection {self.endpoint_uri}"
diff --git a/web3/providers/rpc.py b/web3/providers/rpc.py
index aba4138e..bffa6939 100644
--- a/web3/providers/rpc.py
+++ b/web3/providers/rpc.py
@@ -19,7 +19,7 @@ from web3._utils.http import (
     construct_user_agent,
 )
 from web3._utils.request import (
-    cache_session,
+    cache_and_return_session,
     get_default_http_endpoint,
     make_post_request,
 )
@@ -62,7 +62,7 @@ class HTTPProvider(JSONBaseProvider):
         self._request_kwargs = request_kwargs or {}
 
         if session:
-            cache_session(self.endpoint_uri, session)
+            cache_and_return_session(self.endpoint_uri, session)
 
         super().__init__()
 


Test patch:
diff --git a/tests/core/providers/test_async_http_provider.py b/tests/core/providers/test_async_http_provider.py
index 111e24bb..c5c3b6cc 100644
--- a/tests/core/providers/test_async_http_provider.py
+++ b/tests/core/providers/test_async_http_provider.py
@@ -14,8 +14,8 @@ from web3.providers.async_rpc import (
 
 @pytest.mark.asyncio
 async def test_user_provided_session() -> None:
-
     session = ClientSession()
     provider = AsyncHTTPProvider(endpoint_uri="http://mynode.local:8545")
-    await provider.cache_async_session(session)
+    cached_session = await provider.cache_async_session(session)
     assert len(request._async_session_cache) == 1
+    assert cached_session == session
diff --git a/tests/core/providers/test_http_provider.py b/tests/core/providers/test_http_provider.py
index 247053d0..0e6e15b5 100644
--- a/tests/core/providers/test_http_provider.py
+++ b/tests/core/providers/test_http_provider.py
@@ -38,7 +38,7 @@ def test_user_provided_session():
     w3 = Web3(provider)
     assert w3.manager.provider == provider
 
-    session = request.get_session(URI)
+    session = request.cache_and_return_session(URI)
     adapter = session.get_adapter(URI)
     assert isinstance(adapter, HTTPAdapter)
     assert adapter._pool_connections == 20
diff --git a/tests/core/utilities/test_request.py b/tests/core/utilities/test_request.py
index a1f60112..56c405c8 100644
--- a/tests/core/utilities/test_request.py
+++ b/tests/core/utilities/test_request.py
@@ -1,8 +1,16 @@
+import asyncio
+from concurrent.futures import (
+    ThreadPoolExecutor,
+)
 import pytest
+import time
 
 from aiohttp import (
     ClientSession,
 )
+from eth_typing import (
+    URI,
+)
 from requests import (
     Session,
     adapters,
@@ -15,8 +23,13 @@ from requests.adapters import (
 from web3._utils import (
     request,
 )
+from web3._utils.caching import (
+    generate_cache_key,
+)
 from web3._utils.request import (
     SessionCache,
+    cache_and_return_async_session,
+    cache_and_return_session,
 )
 
 
@@ -34,7 +47,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):
@@ -48,19 +68,23 @@ def test_make_post_request_no_args(mocker):
 
     # Submit a first request to create a session with default parameters
     assert len(request._session_cache) == 0
-    response = request.make_post_request(URI, data=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(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())
@@ -70,44 +94,30 @@ def test_precached_session(mocker):
     session = Session()
     session.mount("http://", adapter)
     session.mount("https://", adapter)
-    request.cache_session(URI, session)
+    request.cache_and_return_session(TEST_URI, session)
 
     # Submit a second request with different arguments
     assert len(request._session_cache) == 1
-    response = request.make_post_request(URI, data=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.cache_and_return_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 +136,129 @@ 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._async_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
+
+    def _simulate_call(uri):
+        _session = cache_and_return_session(uri)
+
+        # simulate a call taking 0.01s to return a response
+        time.sleep(0.01)
+        return _session
+
+    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._async_session_cache = session_cache_default
+    request.DEFAULT_TIMEOUT = timeout_default
+
+    # clear cache
+    request._session_cache.clear()
+
+
+# -- async -- #
+
+
+@pytest.mark.asyncio
+async def test_async_precached_session():
+    # Add a session
+    session = ClientSession()
+    await request.cache_and_return_async_session(TEST_URI, session)
+    assert len(request._async_session_cache) == 1
+
+    # Make sure the session isn't duplicated
+    await request.cache_and_return_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_and_return_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_cache_does_not_close_session_before_a_call_when_multithreading():
+    # save default values
+    session_cache_default = request._async_session_cache
+    timeout_default = request.DEFAULT_TIMEOUT
+
+    # set cache size to 1 + set future session close thread time to 0.01s
+    request._async_session_cache = SessionCache(1)
+    _timeout_for_testing = 0.01
+    request.DEFAULT_TIMEOUT = _timeout_for_testing
+
+    async def cache_uri_and_return_session(uri):
+        _session = await cache_and_return_async_session(uri)
+
+        # simulate a call taking 0.01s to return a response
+        await asyncio.sleep(0.01)
+
+        assert not _session.closed
+        return _session
+
+    tasks = [cache_uri_and_return_session(uri) for uri in UNIQUE_URIS]
+
+    all_sessions = await asyncio.gather(*tasks)
+    assert len(all_sessions) == len(UNIQUE_URIS)
+    assert all(isinstance(s, ClientSession) for s in all_sessions)
+
+    # last session remains in cache, all others evicted
+    cache_data = request._async_session_cache._data
+    assert len(cache_data) == 1
+    _key, cached_session = cache_data.popitem()
+    assert cached_session == all_sessions[-1]
+
+    # assert all evicted sessions were closed
+    await asyncio.sleep(_timeout_for_testing + 0.1)
+    assert all(session.closed for session in all_sessions[:-1])
+
+    # -- teardown -- #
+
+    # appropriately close the cached session
+    await cached_session.close()
+
+    # reset default values
+    request._async_session_cache = session_cache_default
+    request.DEFAULT_TIMEOUT = timeout_default
+
+    # clear cache
+    request._async_session_cache.clear()
diff --git a/web3/_utils/module_testing/module_testing_utils.py b/web3/_utils/module_testing/module_testing_utils.py
index cbf86cbe..9b5a6bab 100644
--- a/web3/_utils/module_testing/module_testing_utils.py
+++ b/web3/_utils/module_testing/module_testing_utils.py
@@ -26,8 +26,8 @@ from web3._utils.compat import (
     Literal,
 )
 from web3._utils.request import (
-    get_async_session,
-    get_session,
+    cache_and_return_async_session,
+    cache_and_return_session,
 )
 from web3.types import (
     BlockData,
@@ -103,7 +103,7 @@ def mock_offchain_lookup_request_response(
             return MockedResponse()
 
         # else, make a normal request (no mocking)
-        session = get_session(url_from_args)
+        session = cache_and_return_session(url_from_args)
         return session.request(method=http_method.upper(), url=url_from_args, **kwargs)
 
     monkeypatch.setattr(
@@ -153,7 +153,7 @@ def async_mock_offchain_lookup_request_response(
             return AsyncMockedResponse()
 
         # else, make a normal request (no mocking)
-        session = await get_async_session(url_from_args)
+        session = await cache_and_return_async_session(url_from_args)
         return await session.request(
             method=http_method.upper(), url=url_from_args, **kwargs
         )