Allow task-based subscription processing via the subscription manager

- For high-throughput subscriptions, where order of processing is not important,
  we should allow a configuration to be set to process tasks in a task-based
  manner - creating a task for each subscription and processing them in parallel.

- Add ``task_based`` configuration to the subscription manager, defaulting to
  ``False``, but allowing this to be set to true to enable task-based processing.

Author: fselmo

web3/_utils/module_testing/persistent_connection_provider.py

@@ -3,6 +3,7 @@
 from dataclasses import (
     dataclass,
 )
+import time
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -15,6 +16,7 @@
 )
 
 from eth_typing import (
+    Address,
     ChecksumAddress,
     HexStr,
 )
@@ -38,6 +40,9 @@
 from web3.middleware import (
     ExtraDataToPOAMiddleware,
 )
+from web3.providers.persistent.request_processor import (
+    TaskReliantQueue,
+)
 from web3.types import (
     BlockData,
     FormattedEthSubscriptionResponse,
@@ -914,3 +919,60 @@ async def test_run_forever_starts_with_0_subs_and_runs_until_task_cancelled(
 
         # cleanup
         await clean_up_task(run_forever_task)
+
+    @pytest.mark.asyncio
+    async def test_high_throughput_subscription_task_based(
+        self, async_w3: AsyncWeb3
+    ) -> None:
+        async_w3.provider._request_processor._handler_subscription_queue = (
+            TaskReliantQueue(maxsize=5_000)
+        )
+        sub_manager = async_w3.subscription_manager
+        sub_manager.task_based = True  # turn on task-based processing
+
+        class Counter:
+            val: int = 0
+
+        counter = Counter()
+
+        async def high_throughput_handler(
+            handler_context: LogsSubscriptionContext,
+        ) -> None:
+            handler_context.counter.val += 1
+            if handler_context.counter.val == 5_000:
+                await handler_context.subscription.unsubscribe()
+            # if we awaited all 5_000 messages, we would sleep at least 5 seconds
+            await asyncio.sleep(5 // 5_000)
+
+        high_throughput_subscription = LogsSubscription(
+            # build a meaningless subscription since we are fabricating the messages
+            Address(b"\x00" * 20),
+            handler=high_throughput_handler,
+            handler_context={"counter": counter},
+        )
+        sub_id = await sub_manager.subscribe(high_throughput_subscription)
+        async_w3.provider._request_processor.cache_request_information(
+            request_id=sub_id,
+            method=RPCEndpoint("eth_subscribe"),
+            params=[],
+            response_formatters=((), (), ()),  # type: ignore
+        )
+
+        # put 5_000 messages in the queue
+        for _ in range(5_000):
+            async_w3.provider._request_processor._handler_subscription_queue.put_nowait(
+                {
+                    "jsonrpc": "2.0",
+                    "method": "eth_subscription",
+                    "params": {"subscription": sub_id, "result": {"foo": "bar"}},  # type: ignore  # noqa: E501
+                }
+            )
+
+        start = time.time()
+        await sub_manager.handle_subscriptions()
+        stop = time.time()
+
+        assert counter.val == 5_000
+
+        assert sub_manager.total_handler_calls == 5_000
+        assert stop - start < 3, "subscription handling took too long!"

web3/providers/persistent/subscription_manager.py

@@ -57,6 +57,9 @@ def __init__(self, w3: "AsyncWeb3") -> None:
         self._provider = cast("PersistentConnectionProvider", w3.provider)
         self._subscription_container = SubscriptionContainer()
 
+        # turn on if order of subscription processing is not important
+        self.task_based = False
+
         # share the subscription container with the request processor so it can separate
         # subscriptions into different queues based on ``sub._handler`` presence
         self._provider._request_processor._subscription_container = (
@@ -281,14 +284,17 @@ async def handle_subscriptions(self, run_forever: bool = False) -> None:
                     sub_id
                 )
                 if sub:
-                    await sub._handler(
-                        EthSubscriptionContext(
-                            self._w3,
-                            sub,
-                            formatted_sub_response["result"],
-                            **sub._handler_context,
-                        )
+                    sub_context = EthSubscriptionContext(
+                        self._w3,
+                        sub,
+                        formatted_sub_response["result"],
+                        **sub._handler_context,
                     )
+                    if self.task_based:
+                        asyncio.create_task(sub._handler(sub_context))
+                    else:
+                        await sub._handler(sub_context)
+
             except SubscriptionProcessingFinished:
                 if not run_forever:
                     self.logger.info(