Fix streaming download performance issue (#224)

2.12.915 (2024-04-02)
=====================

- Fixed a performance issue when streaming download by chunk (size) not
in phase with incoming packets size.
  See jawah/niquests#236

Author: Ousret

CHANGES.rst

@@ -1,3 +1,9 @@
+2.12.915 (2024-04-02)
+=====================
+
+- Fixed a performance issue when streaming download by chunk (size) not in phase with incoming packets size.
+  See https://github.com/jawah/niquests/issues/236
+
 2.12.914 (2024-03-30)
 =====================
 

docker-compose.win.yaml

@@ -1,6 +1,6 @@
 services:
   proxy:
-    image: traefik:v3.3.4-windowsservercore-ltsc2022
+    image: traefik:v3.4.0-rc1-windowsservercore-ltsc2022
     restart: unless-stopped
     depends_on:
       httpbin:

docker-compose.yaml

@@ -1,6 +1,6 @@
 services:
   proxy:
-    image: traefik:v3.3.4
+    image: traefik:v3.4.0-rc1
     restart: unless-stopped
     depends_on:
       httpbin:
@@ -86,8 +86,10 @@ services:
       - ./traefik:/usr/local/etc/haproxy
 
   httpbin:
-    image: mccutchen/go-httpbin:v2.16.1
+    image: mccutchen/go-httpbin:v2.17.0
     restart: unless-stopped
+    deploy:
+      replicas: 6
     labels:
       - traefik.enable=true
       - traefik.http.routers.httpbin-http.rule=Host(`httpbin.local`) || Host(`alt.httpbin.local`)

src/urllib3/_async/response.py

@@ -20,8 +20,8 @@
     ResponseNotReady,
     SSLError,
 )
-from ..response import BytesQueueBuffer, ContentDecoder, HTTPResponse
-from ..util.response import is_fp_closed
+from ..response import ContentDecoder, HTTPResponse
+from ..util.response import is_fp_closed, BytesQueueBuffer
 from ..util.retry import Retry
 from .connection import AsyncHTTPConnection
 

src/urllib3/_version.py

@@ -1,4 +1,4 @@
 # This file is protected via CODEOWNERS
 from __future__ import annotations
 
-__version__ = "2.12.914"
+__version__ = "2.12.915"

src/urllib3/backend/_async/_base.py

@@ -5,6 +5,7 @@
 from ..._collections import HTTPHeaderDict
 from ...contrib.ssa import AsyncSocket, SSLAsyncSocket
 from .._base import BaseBackend, ResponsePromise
+from ...util.response import BytesQueueBuffer
 
 
 class AsyncDirectStreamAccess:
@@ -176,7 +177,7 @@ def __init__(
 
         self._stream_id = stream_id
 
-        self.__buffer_excess: bytes = b""
+        self.__buffer_excess: BytesQueueBuffer = BytesQueueBuffer()
         self.__promise: ResponsePromise | None = None
         self._dsa = dsa
         self._stream_abort = stream_abort
@@ -226,35 +227,24 @@ async def read(self, __size: int | None = None) -> bytes:
                 __size, self._stream_id
             )
 
-            # that's awkward, but rather no choice. the state machine
-            # consume and render event regardless of your amt !
-            if self.__buffer_excess:
-                data = (  # Defensive: Difficult to put in place a scenario that verify this
-                    self.__buffer_excess + data
-                )
-                self.__buffer_excess = b""  # Defensive:
-        else:
-            if __size is None:
-                data = self.__buffer_excess
-                self.__buffer_excess = b""
-            else:
-                data = self.__buffer_excess[:__size]
-                self.__buffer_excess = self.__buffer_excess[__size:]
-
-        if __size is not None and (0 < __size < len(data)):
-            self.__buffer_excess = data[__size:]
-            data = data[:__size]
-
-        if self._eot and len(self.__buffer_excess) == 0:
-            self._stream_abort = None
-            self.closed = True
+            self.__buffer_excess.put(data)
+
+        buf_capacity = len(self.__buffer_excess)
+        data = self.__buffer_excess.get(
+            __size if __size is not None and __size > 0 else buf_capacity
+        )
 
         size_in = len(data)
 
+        buf_capacity -= size_in
+
+        if self._eot and buf_capacity == 0:
+            self._stream_abort = None
+            self.closed = True
+            self._sock = None
+
         if self.chunked:
-            self.chunk_left = (
-                len(self.__buffer_excess) if self.__buffer_excess else None
-            )
+            self.chunk_left = buf_capacity if buf_capacity else None
         elif self.length is not None:
             self.length -= size_in
 

src/urllib3/backend/_base.py

@@ -16,6 +16,7 @@
 
 from .._collections import HTTPHeaderDict
 from .._constant import DEFAULT_BLOCKSIZE, DEFAULT_KEEPALIVE_DELAY
+from ..util.response import BytesQueueBuffer
 
 
 class HttpVersion(str, enum.Enum):
@@ -265,7 +266,7 @@ def __init__(
 
         self._stream_id = stream_id
 
-        self.__buffer_excess: bytes = b""
+        self.__buffer_excess: BytesQueueBuffer = BytesQueueBuffer()
         self.__promise: ResponsePromise | None = None
 
         self.trailers: HTTPHeaderDict | None = None
@@ -337,36 +338,24 @@ def read(self, __size: int | None = None) -> bytes:
                 __size, self._stream_id
             )
 
-            # that's awkward, but rather no choice. the state machine
-            # consume and render event regardless of your amt !
-            if self.__buffer_excess:
-                data = (  # Defensive: Difficult to put in place a scenario that verify this
-                    self.__buffer_excess + data
-                )
-                self.__buffer_excess = b""  # Defensive:
-        else:
-            if __size is None:
-                data = self.__buffer_excess
-                self.__buffer_excess = b""
-            else:
-                data = self.__buffer_excess[:__size]
-                self.__buffer_excess = self.__buffer_excess[__size:]
-
-        if __size is not None and (0 < __size < len(data)):
-            self.__buffer_excess = data[__size:]
-            data = data[:__size]
-
-        if self._eot and len(self.__buffer_excess) == 0:
+            self.__buffer_excess.put(data)
+
+        buf_capacity = len(self.__buffer_excess)
+        data = self.__buffer_excess.get(
+            __size if __size is not None and __size > 0 else buf_capacity
+        )
+
+        size_in = len(data)
+
+        buf_capacity -= size_in
+
+        if self._eot and buf_capacity == 0:
             self._stream_abort = None
             self.closed = True
             self._sock = None
 
-        size_in = len(data)
-
         if self.chunked:
-            self.chunk_left = (
-                len(self.__buffer_excess) if self.__buffer_excess else None
-            )
+            self.chunk_left = buf_capacity if buf_capacity else None
         elif self.length is not None:
             self.length -= size_in
 

src/urllib3/response.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import collections
 import io
 import json as _json
 import logging
@@ -50,7 +49,7 @@
     ResponseNotReady,
     SSLError,
 )
-from .util.response import is_fp_closed
+from .util.response import is_fp_closed, BytesQueueBuffer
 from .util.retry import Retry
 
 if typing.TYPE_CHECKING:
@@ -221,65 +220,6 @@ def _get_decoder(mode: str) -> ContentDecoder:
     return DeflateDecoder()
 
 
-class BytesQueueBuffer:
-    """Memory-efficient bytes buffer
-
-    To return decoded data in read() and still follow the BufferedIOBase API, we need a
-    buffer to always return the correct amount of bytes.
-
-    This buffer should be filled using calls to put()
-
-    Our maximum memory usage is determined by the sum of the size of:
-
-     * self.buffer, which contains the full data
-     * the largest chunk that we will copy in get()
-
-    The worst case scenario is a single chunk, in which case we'll make a full copy of
-    the data inside get().
-    """
-
-    def __init__(self) -> None:
-        self.buffer: typing.Deque[bytes] = collections.deque()
-        self._size: int = 0
-
-    def __len__(self) -> int:
-        return self._size
-
-    def put(self, data: bytes) -> None:
-        self.buffer.append(data)
-        self._size += len(data)
-
-    def get(self, n: int) -> bytes:
-        if n == 0:
-            return b""
-        elif not self.buffer:
-            raise RuntimeError("buffer is empty")
-        elif n < 0:
-            raise ValueError("n should be > 0")
-
-        fetched = 0
-        ret = io.BytesIO()
-        while fetched < n:
-            remaining = n - fetched
-            chunk = self.buffer.popleft()
-            chunk_length = len(chunk)
-            if remaining < chunk_length:
-                left_chunk, right_chunk = chunk[:remaining], chunk[remaining:]
-                ret.write(left_chunk)
-                self.buffer.appendleft(right_chunk)
-                self._size -= remaining
-                break
-            else:
-                ret.write(chunk)
-                self._size -= chunk_length
-            fetched += chunk_length
-
-            if not self.buffer:
-                break
-
-        return ret.getvalue()
-
-
 class HTTPResponse(io.IOBase):
     """
     HTTP Response container.

src/urllib3/util/response.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import collections
+import io
 import re
 import typing
 
@@ -44,3 +46,62 @@ def parse_alt_svc(value: str) -> typing.Iterable[tuple[str, str]]:
     )
 
     yield from re.findall(pattern, value)
+
+
+class BytesQueueBuffer:
+    """Memory-efficient bytes buffer
+
+    To return decoded data in read() and still follow the BufferedIOBase API, we need a
+    buffer to always return the correct amount of bytes.
+
+    This buffer should be filled using calls to put()
+
+    Our maximum memory usage is determined by the sum of the size of:
+
+     * self.buffer, which contains the full data
+     * the largest chunk that we will copy in get()
+
+    The worst case scenario is a single chunk, in which case we'll make a full copy of
+    the data inside get().
+    """
+
+    def __init__(self) -> None:
+        self.buffer: typing.Deque[bytes] = collections.deque()
+        self._size: int = 0
+
+    def __len__(self) -> int:
+        return self._size
+
+    def put(self, data: bytes) -> None:
+        self.buffer.append(data)
+        self._size += len(data)
+
+    def get(self, n: int) -> bytes:
+        if n == 0:
+            return b""
+        elif not self.buffer:
+            raise RuntimeError("buffer is empty")
+        elif n < 0:
+            raise ValueError("n should be > 0")
+
+        fetched = 0
+        ret = io.BytesIO()
+        while fetched < n:
+            remaining = n - fetched
+            chunk = self.buffer.popleft()
+            chunk_length = len(chunk)
+            if remaining < chunk_length:
+                left_chunk, right_chunk = chunk[:remaining], chunk[remaining:]
+                ret.write(left_chunk)
+                self.buffer.appendleft(right_chunk)
+                self._size -= remaining
+                break
+            else:
+                ret.write(chunk)
+                self._size -= chunk_length
+            fetched += chunk_length
+
+            if not self.buffer:
+                break
+
+        return ret.getvalue()