From b668954eed8983b8aedfaf89695e94fcfc5bf156 Mon Sep 17 00:00:00 2001 From: Jens Jorgensen Date: Wed, 11 Nov 2026 17:42:17 -0600 Subject: [PATCH] Split large write buffers into chunks libuv accepts libuv >= 1.52 rejects uv_write()/uv_try_write() requests carrying more than 1024*1024 buffers with UV_EINVAL. A transport that keeps receiving write() calls while libuv still has pending data accumulates every buffer in UVStream._buffer and flushed them in a single request, so a burst of more than 1M small writes in one loop iteration killed the connection with OSError(22, 'Invalid argument'). Hand the buffered writes to libuv in chunks of at most 1024*1024 buffers, one uv_write() request each. --- tests/test_tcp.py | 61 +++++++++++++++++++++++++++++++++++++++ uvloop/handles/stream.pxd | 1 + uvloop/handles/stream.pyx | 52 +++++++++++++++++++++++++++++++++ 3 files changed, 114 insertions(+) diff --git a/tests/test_tcp.py b/tests/test_tcp.py index b933166b..6b36f20c 100644 --- a/tests/test_tcp.py +++ b/tests/test_tcp.py @@ -1190,6 +1190,67 @@ async def run(): self.loop.run_until_complete(run()) + def test_many_queued_writes(self): + # libuv rejects write requests with more than 1024*1024 buffers, + # so the transport has to hand them over in chunks. + N = 1024 * 1024 + 1 + TOTAL = 0 + SENT = 0 + + fut = self.loop.create_future() + + async def server(reader, writer): + nonlocal TOTAL + while True: + d = await reader.read(1024 * 1024) + if not d: + break + TOTAL += len(d) + fut.set_result(True) + writer.close() + + async def run(): + nonlocal SENT + + srv = await asyncio.start_server( + server, + '127.0.0.1', 0, + family=socket.AF_INET) + + addr = srv.sockets[0].getsockname() + r, w = await asyncio.open_connection(*addr) + + # Too many buffers for one request while libuv has nothing + # queued yet. + w.writelines([b'x'] * N) + SENT += N + + # Fill the kernel socket buffer so that libuv has data queued + # and every following write() is buffered by the transport. + BIG = b'y' * (1024 * 1024) + while (w.transport.get_write_buffer_size() == 0 and + not w.transport.is_closing()): + w.write(BIG) + SENT += len(BIG) + + # Too many buffers for one request, flushed at the end of + # this loop iteration. + for _ in range(N): + w.write(b'z') + SENT += N + + await w.drain() + w.close() + await fut + await self.wait_closed(w) + + srv.close() + await srv.wait_closed() + + self.assertEqual(TOTAL, SENT) + + self.loop.run_until_complete(run()) + def test_tcp_handle_abort_in_connection_made(self): async def server(reader, writer): try: diff --git a/uvloop/handles/stream.pxd b/uvloop/handles/stream.pxd index 71c8288e..d3a551d3 100644 --- a/uvloop/handles/stream.pxd +++ b/uvloop/handles/stream.pxd @@ -48,6 +48,7 @@ cdef class UVStream(UVBaseTransport): # _exec_write() is the method that does the actual send, and _try_write() # is a fast-path used in _exec_write() to send a single chunk. cdef inline bint _exec_write(self) except -1 + cdef bint _exec_write_chunks(self) except -1 cdef inline Py_ssize_t _try_write(self, object data) except -2 cdef _close(self) diff --git a/uvloop/handles/stream.pyx b/uvloop/handles/stream.pyx index f8c7f694..fb76cf3c 100644 --- a/uvloop/handles/stream.pyx +++ b/uvloop/handles/stream.pyx @@ -1,6 +1,14 @@ cdef enum: __PREALLOCED_BUFS = 4 + # libuv rejects uv_write()/uv_try_write() requests with more than + # 1024*1024 buffers (UV_EINVAL, see uv__check_before_write() in + # libuv >= 1.52). A transport that keeps receiving write() calls + # while libuv still has pending data for it can easily accumulate + # more than that in a single loop iteration, so UVStream._exec_write() + # hands the buffered data to libuv in chunks of at most this many. + __MAX_WRITE_BUFS = 1024 * 1024 + @cython.no_gc_clear @cython.freelist(DEFAULT_FREELIST_SIZE) @@ -471,6 +479,14 @@ cdef class UVStream(UVBaseTransport): if not buf_len: return True + if buf_len > __MAX_WRITE_BUFS: + # Too many buffers for one libuv request: queue all but the + # last chunk with separate uv_write() calls first. What's + # left in self._buffer is written below as usual. + if not self._exec_write_chunks(): + return False + buf_len = len(self._buffer) + if (self._handle).write_queue_size == 0: # libuv internal write buffers for this stream are empty. if buf_len == 1: @@ -581,6 +597,42 @@ cdef class UVStream(UVBaseTransport): self._maybe_resume_protocol() return False + cdef bint _exec_write_chunks(self) except -1: + # Queue all but the last (up to) __MAX_WRITE_BUFS buffers of + # self._buffer with libuv, one uv_write() request per chunk of + # __MAX_WRITE_BUFS buffers. Returns False if a fatal error + # occurred (self._fatal_error was called), True otherwise. + cdef: + int err + Py_ssize_t pos = 0 + Py_ssize_t buf_len = len(self._buffer) + list chunk + _StreamWriteContext ctx + + while buf_len - pos > __MAX_WRITE_BUFS: + chunk = self._buffer[pos:pos + __MAX_WRITE_BUFS] + pos += __MAX_WRITE_BUFS + + ctx = _StreamWriteContext.new(self, chunk) + err = uv.uv_write(&ctx.req, + self._handle, + ctx.uv_bufs_start, + ctx.uv_bufs_len, + __uv_stream_on_write) + if err < 0: + ctx.close() + self._buffer = [] + self._buffer_size = 0 + exc = convert_error(err) + self._fatal_error(exc, True) + return False + + self._buffer = self._buffer[pos:] + self._buffer_size = 0 + for buf in self._buffer: + self._buffer_size += len(buf) + return True + cdef size_t _get_write_buffer_size(self): if self._handle is NULL: return 0