Standardize HTTPResponse.read(X) behavior regardless of compression
MODIFY
.github/workflows/ci.yml
.github/workflows/ci.yml
@@ -66,11 +66,11 @@ jobs:
os: ubuntu-20.04 # CPython 3.9.2 is not available for ubuntu-22.04.
experimental: false
nox-session: test-3.9
- - python-version: 3.11-dev
+ - python-version: 3.11
os: ubuntu-20.04
experimental: true
nox-session: test-3.11
- - python-version: 3.11-dev
+ - python-version: 3.11
os: ubuntu-22.04
experimental: true
nox-session: test-3.11

ADD
changelog/2128.removal.rst
+Standardized :meth:`~urllib3.response.HTTPResponse.read` to respect the semantics of BufferedIOBase regardless of compression. Specifically, this method:
+
+* only returns an empty bytes object to indicate EOF (that is, the response has been fully consumed),
+* never returns more bytes than requested,
+* can issue any number of system calls: zero, one or multiple.
+
+If you want each :meth:`~urllib3.response.HTTPResponse.read` call to issue a single system call, you need to disable decompression by setting ``decode_content=False``.

MODIFY
dev-requirements.txt
dev-requirements.txt
coverage==6.4
tornado==6.2
PySocks==1.7.1
-pytest==7.0.0
+pytest==7.2.0
pytest-timeout==2.1.0
pytest-freezegun==0.4.2
flaky==3.7.0
@@ -9,3 +9,4 @@ trustme==0.9.0
cryptography==37.0.2
backports.zoneinfo==0.2.1;python_version<"3.9"
towncrier==21.9.0
+pytest-memray==1.3.0;python_version>="3.8" and sys_platform!="win32" and implementation_name=="cpython"

MODIFY
noxfile.py
noxfile.py
import os
import shutil
import subprocess
+import sys
import nox
@@ -31,6 +32,12 @@ def tests_impl(
# Print OpenSSL information.
session.run("python", "-m", "OpenSSL.debug")
+ memray_supported = True
+ if sys.implementation.name != "cpython" or sys.version_info < (3, 8):
+ memray_supported = False # pytest-memray requires CPython 3.8+
+ elif sys.platform == "win32":
+ memray_supported = False
+
# Inspired from https://hynek.me/articles/ditch-codecov-python/
# We use parallel mode and then combine in a later CI step
session.run(
@@ -42,6 +49,7 @@ def tests_impl(
"--parallel-mode",
"-m",
"pytest",
+ *("--memray", "--hide-memray-summary") if memray_supported else (),
"-r",
"a",
f"--color={'yes' if 'GITHUB_ACTIONS' in os.environ else 'auto'}",

MODIFY
src/urllib3/response.py
src/urllib3/response.py
+import collections
import io
import json as _json
import logging
@@ -11,6 +12,7 @@ from socket import timeout as SocketTimeout
from typing import (
TYPE_CHECKING,
Any,
+ Deque,
Generator,
Iterator,
List,
@@ -223,6 +225,63 @@ 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: 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 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 BaseHTTPResponse(io.IOBase):
CONTENT_DECODERS = ["gzip", "deflate"]
if brotli is not None:
@@ -512,6 +571,9 @@ class HTTPResponse(BaseHTTPResponse):
# Determine length of response
self.length_remaining = self._init_length(request_method)
+ # Used to return the correct amount of bytes for partial read()s
+ self._decoded_buffer = BytesQueueBuffer()
+
# If requested, preload the body.
if preload_content and not self._body:
self._body = self.read(decode_content=decode_content)
@@ -720,51 +782,22 @@ class HTTPResponse(BaseHTTPResponse):
# StringIO doesn't like amt=None
return self._fp.read(amt) if amt is not None else self._fp.read()
- def read(
+ def _raw_read(
self,
amt: Optional[int] = None,
- decode_content: Optional[bool] = None,
- cache_content: bool = False,
) -> bytes:
"""
- Similar to :meth:`http.client.HTTPResponse.read`, but with two additional
- parameters: ``decode_content`` and ``cache_content``.
-
- :param amt:
- How much of the content to read. If specified, caching is skipped
- because it doesn't make sense to cache partial content as the full
- response.
-
- :param decode_content:
- If True, will attempt to decode the body based on the
- 'content-encoding' header.
-
- :param cache_content:
- If True, will save the returned data such that the same result is
- returned despite of the state of the underlying file object. This
- is useful if you want the ``.data`` property to continue working
- after having ``.read()`` the file object. (Overridden if ``amt`` is
- set.)
+ Reads `amt` of bytes from the socket.
"""
- self._init_decoder()
- if decode_content is None:
- decode_content = self.decode_content
-
if self._fp is None:
return None # type: ignore[return-value]
- flush_decoder = False
fp_closed = getattr(self._fp, "closed", False)
with self._error_catcher():
data = self._fp_read(amt) if not fp_closed else b""
- if amt is None:
- flush_decoder = True
- else:
- cache_content = False
- if (
- amt != 0 and not data
- ): # Platform-specific: Buggy versions of Python.
+ if amt is not None and amt != 0 and not data:
+ # Platform-specific: Buggy versions of Python.
# Close the connection when no data is returned
#
# This is redundant to what httplib/http.client _should_
@@ -773,7 +806,6 @@ class HTTPResponse(BaseHTTPResponse):
# not properly close the connection in all cases. There is
# no harm in redundantly calling close.
self._fp.close()
- flush_decoder = True
if (
self.enforce_content_length
and self.length_remaining is not None
@@ -790,11 +822,75 @@ class HTTPResponse(BaseHTTPResponse):
self._fp_bytes_read += len(data)
if self.length_remaining is not None:
self.length_remaining -= len(data)
+ return data
- data = self._decode(data, decode_content, flush_decoder)
+ def read(
+ self,
+ amt: Optional[int] = None,
+ decode_content: Optional[bool] = None,
+ cache_content: bool = False,
+ ) -> bytes:
+ """
+ Similar to :meth:`http.client.HTTPResponse.read`, but with two additional
+ parameters: ``decode_content`` and ``cache_content``.
+
+ :param amt:
+ How much of the content to read. If specified, caching is skipped
+ because it doesn't make sense to cache partial content as the full
+ response.
+
+ :param decode_content:
+ If True, will attempt to decode the body based on the
+ 'content-encoding' header.
+
+ :param cache_content:
+ If True, will save the returned data such that the same result is
+ returned despite of the state of the underlying file object. This
+ is useful if you want the ``.data`` property to continue working
+ after having ``.read()`` the file object. (Overridden if ``amt`` is
+ set.)
+ """
+ self._init_decoder()
+ if decode_content is None:
+ decode_content = self.decode_content
+
+ if amt is not None:
+ cache_content = False
+
+ if len(self._decoded_buffer) >= amt:
+ return self._decoded_buffer.get(amt)
+
+ data = self._raw_read(amt)
+
+ flush_decoder = False
+ if amt is None:
+ flush_decoder = True
+ elif amt != 0 and not data:
+ flush_decoder = True
+
+ if not data and len(self._decoded_buffer) == 0:
+ return data
+ if amt is None:
+ data = self._decode(data, decode_content, flush_decoder)
if cache_content:
self._body = data
+ else:
+ # do not waste memory on buffer when not decoding
+ if not decode_content:
+ return data
+
+ decoded_data = self._decode(data, decode_content, flush_decoder)
+ self._decoded_buffer.put(decoded_data)
+
+ while len(self._decoded_buffer) < amt and data:
+ # TODO make sure to initially read enough data to get past the headers
+ # For example, the GZ file header takes 10 bytes, we don't want to read
+ # it one byte at a time
+ data = self._raw_read(amt)
+ decoded_data = self._decode(data, decode_content, flush_decoder)
+ self._decoded_buffer.put(decoded_data)
+ data = self._decoded_buffer.get(amt)
return data

MODIFY
test/test_response.py
test/test_response.py
@@ -2,6 +2,7 @@ import contextlib
import http.client as httplib
import socket
import ssl
+import sys
import zlib
from base64 import b64decode
from http.client import IncompleteRead as httplib_IncompleteRead
@@ -24,6 +25,7 @@ from urllib3.exceptions import (
)
from urllib3.response import ( # type: ignore[attr-defined]
BaseHTTPResponse,
+ BytesQueueBuffer,
HTTPResponse,
brotli,
zstd,
@@ -31,6 +33,56 @@ from urllib3.response import ( # type: ignore[attr-defined]
from urllib3.util.response import is_fp_closed
from urllib3.util.retry import RequestHistory, Retry
+
+class TestBytesQueueBuffer:
+ def test_single_chunk(self) -> None:
+ buffer = BytesQueueBuffer()
+ assert len(buffer) == 0
+ with pytest.raises(RuntimeError, match="buffer is empty"):
+ assert buffer.get(10)
+
+ buffer.put(b"foo")
+ with pytest.raises(ValueError, match="n should be > 0"):
+ buffer.get(-1)
+
+ assert buffer.get(1) == b"f"
+ assert buffer.get(2) == b"oo"
+ with pytest.raises(RuntimeError, match="buffer is empty"):
+ assert buffer.get(10)
+
+ def test_read_too_much(self) -> None:
+ buffer = BytesQueueBuffer()
+ buffer.put(b"foo")
+ assert buffer.get(100) == b"foo"
+
+ def test_multiple_chunks(self) -> None:
+ buffer = BytesQueueBuffer()
+ buffer.put(b"foo")
+ buffer.put(b"bar")
+ buffer.put(b"baz")
+ assert len(buffer) == 9
+
+ assert buffer.get(1) == b"f"
+ assert len(buffer) == 8
+ assert buffer.get(4) == b"ooba"
+ assert len(buffer) == 4
+ assert buffer.get(4) == b"rbaz"
+ assert len(buffer) == 0
+
+ @pytest.mark.skipif(
+ sys.version_info < (3, 8), reason="pytest-memray requires Python 3.8+"
+ )
+ @pytest.mark.limit_memory("12.5 MB") # assert that we're not doubling memory usage
+ def test_memory_usage(self) -> None:
+ # Allocate 10 1MiB chunks
+ buffer = BytesQueueBuffer()
+ for i in range(10):
+ # This allocates 2MiB, putting the max at around 12MiB. Not sure why.
+ buffer.put(bytes(2**20))
+
+ assert len(buffer.get(10 * 2**20)) == 10 * 2**20
+
+
# A known random (i.e, not-too-compressible) payload generated with:
# "".join(random.choice(string.printable) for i in range(512))
# .encode("zlib").encode("base64")
@@ -71,9 +123,19 @@ class TestLegacyResponse:
class TestResponse:
def test_cache_content(self) -> None:
r = HTTPResponse(b"foo")
+ assert r._body == b"foo"
assert r.data == b"foo"
assert r._body == b"foo"
+ def test_cache_content_preload_false(self) -> None:
+ fp = BytesIO(b"foo")
+ r = HTTPResponse(fp, preload_content=False)
+
+ assert not r._body
+ assert r.data == b"foo"
+ assert r._body == b"foo"
+ assert r.data == b"foo"
+
def test_default(self) -> None:
r = HTTPResponse()
assert r.data is None
@@ -137,13 +199,7 @@ class TestResponse:
fp, headers={"content-encoding": "deflate"}, preload_content=False
)
- assert r.read(3) == b""
- # Buffer in case we need to switch to the raw stream
- assert r._decoder is not None
- assert r._decoder._data is not None # type: ignore[attr-defined]
assert r.read(1) == b"f"
- # Now that we've decoded data, we just stream through the decoder
- assert r._decoder._data is None # type: ignore[attr-defined]
assert r.read(2) == b"oo"
assert r.read() == b""
assert r.read() == b""
@@ -158,11 +214,7 @@ class TestResponse:
fp, headers={"content-encoding": "deflate"}, preload_content=False
)
- assert r.read(1) == b""
assert r.read(1) == b"f"
- # Once we've decoded data, we just stream to the decoder; no buffering
- assert r._decoder is not None
- assert r._decoder._data is None # type: ignore[attr-defined]
assert r.read(2) == b"oo"
assert r.read() == b""
assert r.read() == b""
@@ -177,7 +229,6 @@ class TestResponse:
fp, headers={"content-encoding": "gzip"}, preload_content=False
)
- assert r.read(11) == b""
assert r.read(1) == b"f"
assert r.read(2) == b"oo"
assert r.read() == b""
@@ -324,6 +375,23 @@ class TestResponse:
assert r.data == b"foo"
+ def test_read_multi_decoding_deflate_deflate(self) -> None:
+ msg = b"foobarbaz" * 42
+ data = zlib.compress(zlib.compress(msg))
+
+ fp = BytesIO(data)
+ r = HTTPResponse(
+ fp, headers={"content-encoding": "deflate, deflate"}, preload_content=False
+ )
+
+ assert r.read(3) == b"foo"
+ assert r.read(3) == b"bar"
+ assert r.read(3) == b"baz"
+ assert r.read(9) == b"foobarbaz"
+ assert r.read(9 * 3) == b"foobarbaz" * 3
+ assert r.read(9 * 37) == b"foobarbaz" * 37
+ assert r.read() == b""
+
def test_body_blob(self) -> None:
resp = HTTPResponse(b"foo")
assert resp.data == b"foo"
@@ -527,8 +595,8 @@ class TestResponse:
)
stream = resp.stream(2)
- assert next(stream) == b"f"
- assert next(stream) == b"oo"
+ assert next(stream) == b"fo"
+ assert next(stream) == b"o"
with pytest.raises(StopIteration):
next(stream)
@@ -557,6 +625,7 @@ class TestResponse:
# Ensure that ``tell()`` returns the correct number of bytes when
# part-way through streaming compressed content.
NUMBER_OF_READS = 10
+ PART_SIZE = 64
class MockCompressedDataReading(BytesIO):
"""
@@ -585,7 +654,7 @@ class TestResponse:
resp = HTTPResponse(
fp, headers={"content-encoding": "deflate"}, preload_content=False
)
- stream = resp.stream()
+ stream = resp.stream(PART_SIZE)
parts_positions = [(part, resp.tell()) for part in stream]
end_of_stream = resp.tell()
@@ -600,12 +669,28 @@ class TestResponse:
assert uncompressed_data == payload
# Check that the positions in the stream are correct
- expected = [(i + 1) * payload_part_size for i in range(NUMBER_OF_READS)]
- assert expected == list(positions)
+ # It is difficult to determine programatically what the positions
+ # returned by `tell` will be because the `HTTPResponse.read` method may
+ # call socket `read` a couple of times if it doesn't have enough data
+ # in the buffer or not call socket `read` at all if it has enough. All
+ # this depends on the message, how it was compressed, what is
+ # `PART_SIZE` and `payload_part_size`.
+ # So for simplicity the expected values are hardcoded.
+ expected = (92, 184, 230, 276, 322, 368, 414, 460)
+ assert expected == positions
# Check that the end of the stream is in the correct place
assert len(ZLIB_PAYLOAD) == end_of_stream
+ # Check that all parts have expected length
+ expected_last_part_size = len(uncompressed_data) % PART_SIZE
+ whole_parts = len(uncompressed_data) // PART_SIZE
+ if expected_last_part_size == 0:
+ expected_lengths = [PART_SIZE] * whole_parts
+ else:
+ expected_lengths = [PART_SIZE] * whole_parts + [expected_last_part_size]
+ assert expected_lengths == [len(part) for part in parts]
+
def test_deflate_streaming(self) -> None:
data = zlib.compress(b"foo")
@@ -615,8 +700,8 @@ class TestResponse:
)
stream = resp.stream(2)
- assert next(stream) == b"f"
- assert next(stream) == b"oo"
+ assert next(stream) == b"fo"
+ assert next(stream) == b"o"
with pytest.raises(StopIteration):
next(stream)
@@ -631,8 +716,8 @@ class TestResponse:
)
stream = resp.stream(2)
- assert next(stream) == b"f"
- assert next(stream) == b"oo"
+ assert next(stream) == b"fo"
+ assert next(stream) == b"o"
with pytest.raises(StopIteration):
next(stream)
@@ -644,6 +729,38 @@ class TestResponse:
with pytest.raises(StopIteration):
next(stream)
+ @pytest.mark.parametrize(
+ "preload_content, amt",
+ [(True, None), (False, None), (False, 10 * 2**20)],
+ )
+ @pytest.mark.limit_memory("25 MB")
+ def test_buffer_memory_usage_decode_one_chunk(
+ self, preload_content: bool, amt: int
+ ) -> None:
+ content_length = 10 * 2**20 # 10 MiB
+ fp = BytesIO(zlib.compress(bytes(content_length)))
+ resp = HTTPResponse(
+ fp,
+ preload_content=preload_content,
+ headers={"content-encoding": "deflate"},
+ )
+ data = resp.data if preload_content else resp.read(amt)
+ assert len(data) == content_length
+
+ @pytest.mark.parametrize(
+ "preload_content, amt",
+ [(True, None), (False, None), (False, 10 * 2**20)],
+ )
+ @pytest.mark.limit_memory("10.5 MB")
+ def test_buffer_memory_usage_no_decoding(
+ self, preload_content: bool, amt: int
+ ) -> None:
+ content_length = 10 * 2**20 # 10 MiB
+ fp = BytesIO(bytes(content_length))
+ resp = HTTPResponse(fp, preload_content=preload_content, decode_content=False)
+ data = resp.data if preload_content else resp.read(amt)
+ assert len(data) == content_length
+
def test_length_no_header(self) -> None:
fp = BytesIO(b"12345")
resp = HTTPResponse(fp, preload_content=False)

MODIFY
test/with_dummyserver/test_socketlevel.py
test/with_dummyserver/test_socketlevel.py
@@ -1920,15 +1920,8 @@ class TestBadContentLength(SocketDummyServerTestCase):
"GET", url="/", preload_content=False, enforce_content_length=True
)
data = get_response.stream(100)
- # Read "good" data before we try to read again.
- # This won't trigger till generator is exhausted.
+ with pytest.raises(ProtocolError, match="12 bytes read, 10 more expected"):
next(data)
- try:
- next(data)
- assert False
- except ProtocolError as e:
- assert "12 bytes read, 10 more expected" in str(e)
-
done_event.set()
def test_enforce_content_length_no_body(self) -> None:

Standardize HTTPResponse.read(X) behavior regardless of compression

Co-authored-by: Franek Magiera <framagie@gmail.com>