Add benchmarks for streaming API (#10644)

<!-- Thank you for your contribution! -->

## What do these changes do?

Add benchmarks for streaming API.

I get asked all the time if someone should use `iter_chunked`,
`iter_any`, or `iter_chunks`. Its nice to be able to point people at
benchmarks, especially ones they can alter with their numbers and run
locally.

Unsurprisingly `memcpy` is where the performance issues are so avoiding
it as much as possible will give better performance. In almost all cases
`iter_chunks` is going to give the best performance if you can trust the
chunk sizes aren't going to be too large.

## Are there changes in behavior for the user?

no

## Is it a substantial burden for the maintainers to support this?
no

Author: bdraco

tests/test_benchmarks_client.py

@@ -346,3 +346,131 @@ async def run_client_benchmark() -> None:
     @benchmark
     def _run() -> None:
         loop.run_until_complete(run_client_benchmark())
+
+
+def test_ten_streamed_responses_iter_any(
+    loop: asyncio.AbstractEventLoop,
+    aiohttp_client: AiohttpClient,
+    benchmark: BenchmarkFixture,
+) -> None:
+    """Benchmark 10 streamed responses using iter_any."""
+    message_count = 10
+    data = b"x" * 65536  # 64 KiB chunk size
+
+    async def handler(request: web.Request) -> web.StreamResponse:
+        resp = web.StreamResponse()
+        await resp.prepare(request)
+        for _ in range(10):
+            await resp.write(data)
+        return resp
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+
+    async def run_client_benchmark() -> None:
+        client = await aiohttp_client(app)
+        for _ in range(message_count):
+            resp = await client.get("/")
+            async for _ in resp.content.iter_any():
+                pass
+        await client.close()
+
+    @benchmark
+    def _run() -> None:
+        loop.run_until_complete(run_client_benchmark())
+
+
+def test_ten_streamed_responses_iter_chunked_4096(
+    loop: asyncio.AbstractEventLoop,
+    aiohttp_client: AiohttpClient,
+    benchmark: BenchmarkFixture,
+) -> None:
+    """Benchmark 10 streamed responses using iter_chunked 4096."""
+    message_count = 10
+    data = b"x" * 65536  # 64 KiB chunk size, 4096 iter_chunked
+
+    async def handler(request: web.Request) -> web.StreamResponse:
+        resp = web.StreamResponse()
+        await resp.prepare(request)
+        for _ in range(10):
+            await resp.write(data)
+        return resp
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+
+    async def run_client_benchmark() -> None:
+        client = await aiohttp_client(app)
+        for _ in range(message_count):
+            resp = await client.get("/")
+            async for _ in resp.content.iter_chunked(4096):
+                pass
+        await client.close()
+
+    @benchmark
+    def _run() -> None:
+        loop.run_until_complete(run_client_benchmark())
+
+
+def test_ten_streamed_responses_iter_chunked_65536(
+    loop: asyncio.AbstractEventLoop,
+    aiohttp_client: AiohttpClient,
+    benchmark: BenchmarkFixture,
+) -> None:
+    """Benchmark 10 streamed responses using iter_chunked 65536."""
+    message_count = 10
+    data = b"x" * 65536  # 64 KiB chunk size, 64 KiB iter_chunked
+
+    async def handler(request: web.Request) -> web.StreamResponse:
+        resp = web.StreamResponse()
+        await resp.prepare(request)
+        for _ in range(10):
+            await resp.write(data)
+        return resp
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+
+    async def run_client_benchmark() -> None:
+        client = await aiohttp_client(app)
+        for _ in range(message_count):
+            resp = await client.get("/")
+            async for _ in resp.content.iter_chunked(65536):
+                pass
+        await client.close()
+
+    @benchmark
+    def _run() -> None:
+        loop.run_until_complete(run_client_benchmark())
+
+
+def test_ten_streamed_responses_iter_chunks(
+    loop: asyncio.AbstractEventLoop,
+    aiohttp_client: AiohttpClient,
+    benchmark: BenchmarkFixture,
+) -> None:
+    """Benchmark 10 streamed responses using iter_chunks."""
+    message_count = 10
+    data = b"x" * 65536  # 64 KiB chunk size
+
+    async def handler(request: web.Request) -> web.StreamResponse:
+        resp = web.StreamResponse()
+        await resp.prepare(request)
+        for _ in range(10):
+            await resp.write(data)
+        return resp
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+
+    async def run_client_benchmark() -> None:
+        client = await aiohttp_client(app)
+        for _ in range(message_count):
+            resp = await client.get("/")
+            async for _ in resp.content.iter_chunks():
+                pass
+        await client.close()
+
+    @benchmark
+    def _run() -> None:
+        loop.run_until_complete(run_client_benchmark())