Adding locks when manipulating the _session_cache object in case of multi-threading

web3/_utils/request.py

@@ -1,7 +1,14 @@
+import os
+from threading import (
+    RLock,
+)
 from typing import (
     Any,
 )
-
+from aiohttp import (
+    ClientSession,
+    ClientTimeout,
+)
 from eth_typing import (
     URI,
 )
@@ -12,23 +19,59 @@
     generate_cache_key,
 )
 
+_lock = RLock()
+
+
+def _acquireLock() -> None:
+    """
+    Acquire the module-level lock for serializing access to shared cache.
+
+    This should be released with _releaseLock().
+    """
+    if _lock:
+        _lock.acquire()
+
+
+def _releaseLock() -> None:
+    """
+    Release the module-level lock acquired by calling _acquireLock().
+    """
+    if _lock:
+        _lock.release()
+
+
+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
+    _acquireLock()
+    try:
+        cache_key = generate_cache_key(endpoint_uri)
+        _session_cache[cache_key] = session
+    finally:
+        _releaseLock()
 
 
 def _remove_session(key: str, session: requests.Session) -> None:
-    session.close()
+    _acquireLock()
+    try:
+        session.close()
+    finally:
+        _releaseLock()
 
 
 _session_cache = lru.LRU(8, callback=_remove_session)
 
 
 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()
+    _acquireLock()
+    try:
+        cache_key = generate_cache_key(endpoint_uri)
+        if cache_key not in _session_cache:
+            _session_cache[cache_key] = requests.Session()
+    finally:
+        _releaseLock()
     return _session_cache[cache_key]
 
 
@@ -40,3 +83,15 @@ def make_post_request(endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any)
     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))
+    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()