Semaphore (#13)

* chore: shrink default loop interval to 0.001s and move to _config file

* fix: AttributeError

* feat: Semaphore to limit dank_mids congestion, default value 100k

.gitignore

@@ -2,6 +2,8 @@ build
 dist
 env
 dank_mids.egg-info
+.hypothesis
+.pytest_cache
 .eggs
 .mypy_cache
 __pycache__

dank_mids/_config.py

@@ -0,0 +1,6 @@
+
+import asyncio
+import os
+
+SEMAPHORE = asyncio.Semaphore(int(os.environ.get("DANKMIDSSEMAPHORE", 100_000)))
+LOOP_INTERVAL = float(os.environ.get("DANKMIDSLOOPINTERVAL", 0.001))

dank_mids/constants.py

@@ -1,8 +1,6 @@
 
-import os
 import multicall
 
-LOOP_INTERVAL = float(os.environ.get("DANKMIDSLOOPINTERVAL", 0.05))
 GAS_LIMIT = multicall.constants.GAS_LIMIT
 OVERRIDE_CODE = multicall.constants.MULTICALL2_BYTECODE
 

dank_mids/controller.py

@@ -16,8 +16,8 @@
 from web3.providers.async_base import AsyncBaseProvider
 from web3.types import RPCEndpoint, RPCResponse
 
-from dank_mids.constants import (BAD_HEXES, GAS_LIMIT, LOOP_INTERVAL,
-                                 OVERRIDE_CODE)
+from dank_mids._config import LOOP_INTERVAL
+from dank_mids.constants import BAD_HEXES, GAS_LIMIT, OVERRIDE_CODE
 from dank_mids.loggers import (demo_logger, main_logger, sort_lazy_logger,
                                sort_logger)
 
@@ -144,7 +144,7 @@ async def execute_multicall(self) -> None:
         i = 0
         while self._cid_lock.locked():
             if i // 50 == int(i // 50):
-                self.logger.debug('lock is locked')
+                main_logger.debug('lock is locked')
             await asyncio.sleep(.1)
         if self._is_executing:
             return
@@ -196,7 +196,7 @@ async def process_batch(self, batch: List, block: str, bid: Optional[int] = None
                     self.batcher.step = new_step
                     main_logger.warning(f'Multicall batch size reduced from {old_step} to {new_step}. The failed batch had {len(inputs)} calls.')
             else:
-                self.logger.warning(f"unexpected exception: {type(e)} {str(e)}")
+                main_logger.warning(f"unexpected exception: {type(e)} {str(e)}")
 
             batches = [[cid, input] for cid, input in zip(cids, inputs)]
             halfpoint = len(batches) // 2
@@ -256,7 +256,7 @@ async def _setup(self) -> None:
                 await asyncio.sleep(0)
             return
         self._initializing = True
-        self.logger.info('Dank Middleware initializing... Strap on your rocket boots...')
+        main_logger.info('Dank Middleware initializing... Strap on your rocket boots...')
         # NOTE use sync w3 here to prevent timeout issues with abusive scripts.
         chain_id = self.sync_w3.eth.chain_id
         MULTICALL = multicall.constants.MULTICALL_ADDRESSES.get(chain_id,None)

dank_mids/middleware.py

@@ -4,6 +4,7 @@
 from web3 import Web3
 from web3.types import RPCEndpoint, RPCResponse
 
+from dank_mids._config import SEMAPHORE
 from dank_mids.controller import DankMiddlewareController
 from dank_mids.types import AsyncMiddleware
 
@@ -17,6 +18,7 @@ async def middleware(method: RPCEndpoint, params: Any) -> RPCResponse:
         if not dank_mids._is_configured and dank_mids.should_batch(method, params):
             await dank_mids._setup()
         if dank_mids.should_batch(method, params):
-            return await dank_mids(params)
+            async with SEMAPHORE:
+                return await dank_mids(params)
         return await make_request(method, params)
     return middleware