Implement zero copy writes for the WebSocket writer

[bdraco]

Implement zero copy writes for the WebSocket writer
closes #9633

python/cpython#91166 made writelines zero copy for Python 3.12+. For older versions it will still join the bytes

Would like another set of 👀 on this to make sure it makes sense.

[codspeed-hq]

CodSpeed Performance Report

Merging #9634 will degrade performances by 13.08%

_{Comparing zeroconf_writes (6fbb4ff) with master (27e23fb)}

Summary

⚡ 1 improvements
❌ 1 regressions
✅ 40 untouched benchmarks

⚠️ Please fix the performance issues or acknowledge them on CodSpeed.

Benchmarks breakdown

	Benchmark	master	zeroconf_writes	Change
❌	test_one_thousand_large_round_trip_websocket_text_messages[pyloop]	23.5 ms	27 ms	-13.08%
⚡	test_send_one_hundred_websocket_text_messages_with_mask[pyloop]	807.3 µs	746.2 µs	+8.18%

[codecov]

❌ 6 Tests Failed:

Tests completed	Failed	Passed	Skipped
3447	6	3441	16

View the top 3 failed tests by shortest run time

tests.test_client_ws_functional test_heartbeat_connection_closed[pyloop]

Stack Traces | 0.53s run time

No failure message available

tests.test_client_ws_functional test_heartbeat[pyloop]

Stack Traces | 10.1s run time

No failure message available

tests.test_client_ws_functional test_heartbeat_no_pong_concurrent_receive[pyloop]

Stack Traces | 10.2s run time

No failure message available

To view more test analytics, go to the Test Analytics Dashboard
Got feedback? Let us know on Github

[bdraco]

Benchmarks barely show any improvement. I think the message payload would have to be rather large for it to matter.

We don't have any benchmarks for that

[bdraco]

Still probably not a good benchmark. Will need to extend the round trip WebSocket benchmark to have another one that sends large messages to really get an idea of how it will perform

[bdraco]

I think part of the reason it’s slower is the protocol gets paused. Everything backs up against the drain waiter in the writer anyways so maybe we need a queue that doesn’t pause writes in two places

Nevermind. I was misremembering. It’s reads that get paused which makes more sense

[bdraco]

Anyways. Let’s see what the profile looks like with now with all the memcpy reduction optimizations

[bdraco]

We still end up with more fragmentation somehow.

Probably need to work on the reader some more

[bdraco]

need to look at this with syscalls

[bdraco]

I got this working so it was a net win for the StreamWriter in #9839

I think we need better logic to handle larger payloads only with writelines.

Running it again now so I can look at the flame graphs once codspeed finishes as it will likely give a better idea on where the breakpoint is

[bdraco]

The websocket data queue is very small 65k so as soon as we get a large message, we always pause the reader

[asvetlov]

For now, +-10% difference doesn't look very impressive.

[bdraco]

For now, +-10% difference doesn't look very impressive.

no it definitely doesn't

[bdraco]

also codspeed doesn't show syscalls since they can't be consistently instrumented so anything that does syscalls has to be checked manually in the codspeed UI since it can be a significant improvement or significant degradation.

[asvetlov]

@bdraco if you are looking for possible optimizations, replacing https://docs.python.org/3/library/asyncio-protocol.html#asyncio.Protocol with https://docs.python.org/3/library/asyncio-protocol.html#asyncio.BufferedProtocol could be an interesting experiment.
In nutshell, the idea is:
The buffered protocol could use at least two preallocated buffers.
Read from socket into buffer 1, parse buffer 2. After parsing switch them.
It removes data allocation on reading and at least in theory could work well with zero-copy on writing.

[bdraco]

That could help quite a bit if they are small. Not sure if it will help if they are not small though since we have to combine them at some point to wrap it in WSMessage.

Still most messages are small so it should save one memcpy on most messages

[bdraco]

Oh but we still have to do one slice to split of the headers and message body.....hmmm

[asvetlov]

No, I'm talking about a little different situation. Memcpy stays as is, but preallocated membuffers could eliminate alloc/free calls.
Now data_received() allocates bytes, files it with data from the socket, and returns bytes back.
Aiohttp handles it, copies and de-allocates given bytestring.
Buffered protocol could help to avoid these alloc/free.
I don't expect a very big difference though.

[bdraco]

No, I'm talking about a little different situation. Memcpy stays as is, but preallocated membuffers could eliminate alloc/free calls. Now data_received() allocates bytes, files it with data from the socket, and returns bytes back. Aiohttp handles it, copies and de-allocates given bytestring. Buffered protocol could help to avoid these alloc/free.
Got it. I was so focused on the memcpy in the profile that wasn't think about this case.

I don't expect a very big difference though.

Yeah I expect it won't make much difference.

Any idea on how to avoid the copy at

aiohttp/aiohttp/_http_parser.pyx

Line 766 in be31bed

cdef bytes body = at[:length]

?

[asvetlov]

I'm far from my laptop right now, I cannot watch at generated Cython code.
But maybe memoryview could help? Memoryview slicing doesn't perform memcpy but creates a new view which references the same buffer but with different offsets.

[bdraco]

diff --git a/aiohttp/_http_parser.pyx b/aiohttp/_http_parser.pyx
index 30767dae6..804f3f8b5 100644
--- a/aiohttp/_http_parser.pyx
+++ b/aiohttp/_http_parser.pyx
@@ -11,6 +11,7 @@ from cpython cimport (
     PyBytes_AsStringAndSize,
     PyObject_GetBuffer,
 )
+from cpython.memoryview cimport PyMemoryView_FromMemory
 from cpython.mem cimport PyMem_Free, PyMem_Malloc
 from libc.limits cimport ULLONG_MAX
 from libc.string cimport memcpy
@@ -763,7 +764,7 @@ cdef int cb_on_headers_complete(cparser.llhttp_t* parser) except -1:
 cdef int cb_on_body(cparser.llhttp_t* parser,
                     const char *at, size_t length) except -1:
     cdef HttpParser pyparser = <HttpParser>parser.data
-    cdef bytes body = at[:length]
+    body = PyMemoryView_FromMemory(<char*>at, length, 0);
     try:
         pyparser._payload.feed_data(body)
     except BaseException as underlying_exc:

This kinda works but only if we don't release the buffer.

Ideally we find a way for this to defer releasing the buffer until all the memory views are consumed...