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,
@@ -38,6 +39,9 @@
 from web3.middleware import (
     ExtraDataToPOAMiddleware,
 )
+from web3.providers.persistent.request_processor import (
+    TaskReliantQueue,
+)
 from web3.types import (
     BlockData,
     FormattedEthSubscriptionResponse,
@@ -914,3 +918,62 @@ 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
+
+        async def high_throughput_handler(
+            handler_context: LogsSubscriptionContext,
+        ) -> None:
+            handler_context.counter.val += 1
+            if (
+                handler_context.async_w3.subscription_manager.total_handler_calls
+                == 5_000
+            ):
+                await handler_context.subscription.unsubscribe()
+            # 3 seconds / 5_000 = 0.0006 seconds. If we awaited, we would expect this to
+            # take at least 3 seconds to process all 5_000 messages.
+            await asyncio.sleep(0.0006)
+
+        class Counter:
+            val: int = 0
+
+        counter = Counter()
+        high_throughput_subscription = LogsSubscription(
+            "0xdead00000000000000000000000000000000beef",  # type: ignore
+            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=["testestest"],
+            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(