Split large write buffers into chunks libuv accepts

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:

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)

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 (<uv.uv_stream_t*>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,
+                              <uv.uv_stream_t*>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