Add optional executor offloading for client response decompression - #14316

@@           Coverage Diff            @@
##           master   #14316    +/-   ##
========================================
  Coverage   99.11%   99.12%            
========================================
  Files         135      135            
  Lines       53439    54138   +699     
  Branches     2723     2779    +56     
========================================
+ Hits        52967    53664   +697     
- Misses        358      360     +2     
  Partials      114      114            

Flags with carried forward coverage won't be shown. Click here to find out more.

⚠️ 52 benchmarks are dominated by system calls

System calls cannot be simulated consistently and are left out of the measure. The reported value understates the real cost. Switch to the Walltime instrument to measure them.

Learn more about system calls

See affected benchmarks

⚠️ 7 benchmarks with a significant change ran on different hardware

Their base and head results were measured on machines with a different CPU, core count or memory, so the changes might not come from the code. Run base and head on the same runner type.

Learn more about Macro Runners

See affected benchmarks

⚠️ 7 benchmarks with a significant change ran in a different software environment

Their base and head results were measured with a different OS or toolchain, such as compiler, runtime or library versions, so the changes might come from this difference rather than from the code. Open an affected benchmark to see what changed.

See affected benchmarks

• Warming up the performance wizard...
• Handling comment — Comment handling finished successfully

The PR title and description still describe an opt-in auto_decompress_in_executor flag, but the second commit ("Always offload client response decompression") removed that flag from ClientSession, request(), and the docs. At the PR head, set_response_params sets _offload_decompression = auto_decompress, so every compressed response for every user now runs through the loop's default executor, with at least one run_in_executor hop per response plus one per empty-input drain. The only way to keep decoding inline is auto_decompress=False.

Why this matters beyond a stale description:

• DeflateBuffer.feed_data calls protocol.pause_reading() while a worker is in flight, and ResponseHandler.pause_reading() drops the sock_read timer. The timer is only re-armed once reading fully resumes. So the time spent waiting on the default executor is excluded from sock_read. A busy or deliberately small default executor (shared with getaddrinfo, file I/O, user run_in_executor work) turns a stream configured with sock_read but no total into an unbounded wait. Previously the user could at least avoid this by not enabling the feature.
• Small compressed responses (the common JSON/API case) now pay a thread hop and a context switch for a few hundred bytes of inflate, which the inline path did for free. CodSpeed only reports the new benchmark without a baseline, so this cost is not visible in the numbers posted so far.
• docs/client_reference.rst still documents the request-level auto_decompress (lines 586 and 1057) as plain "automatically decompress response body" with no mention of the thread pool; only the session parameter got the versionchanged note.

Suggested path: if always-on is the intended direction, say so in the PR title/description and docs, and give users an escape hatch: either keep a session/request knob, or follow the existing compression_utils convention (MAX_SYNC_CHUNK_SIZE, zlib_executor_size in web_response.py) and only offload when the input chunk exceeds a threshold or when it is an empty-input drain (which implies retained input). If opt-in is still the intent, the second commit needs to be reverted.

        self._offload_decompression = auto_decompress

        self._read_bufsize = read_bufsize
        self._read_timeout = read_timeout

        self._timeout_ceil_threshold = timeout_ceil_threshold

        self._parser = HttpResponseParser(
            self,
            self._loop,
            read_bufsize,
            timer=timer,
            payload_exception=ClientPayloadError,
            response_with_body=not skip_payload,
            read_until_eof=read_until_eof,
            auto_decompress=auto_decompress,
            offload_decompression=auto_decompress,

pending.result() raises asyncio.CancelledError when the executor cancels a queued job (for example executor.shutdown(cancel_futures=True) while a session is still live). That is a BaseException, so it escapes this except Exception, escapes the pure-Python HttpParser.feed_data handler at lines 576-596 (which only catches Exception and is what sets the payload exception), and surfaces in ResponseHandler.data_received. The protocol compensates with a dedicated branch at client_proto.py:425-430, but only when _deferred_eof is set. In the live-transport case the CancelledError is re-raised out of the _decompression_done loop callback (logged by asyncio as "Exception in callback"), the Python-parser payload gets no exception until connection_lost arrives, and the Cython parser (whose cb_on_body catches BaseException) behaves differently.

Catching the cancellation here and converting it to ContentEncodingError would make the two parsers behave identically and let you delete the CancelledError special case in data_received:

except (Exception, asyncio.CancelledError) as exc:
    raise ContentEncodingError(...) from exc

This is a narrow edge (executor shutdown under a live session), so non-blocking.

try:
    decompressed = pending.result()
except Exception as exc:
    raise ContentEncodingError(
        "Can not decode content-encoding: %s" % self.encoding
    ) from exc

This guard, and the pre-drain self.data_received(b"") added to ResponseHandler.connection_lost (client_proto.py:165-170), are keyed on _offload_decompression, which at the PR head equals auto_decompress and is therefore on by default for every response. But _tail on the C parser is not decompression-specific: a plain body paused for StreamReader high water also leaves unconsumed bytes in _tail (cb_on_body returns HPE_PAUSED). So with the default settings a plain-body connection loss on the C parser now returns None here, flips _deferred_eof on, and completes later from the consumer's resume_reading(), while the pure-Python parser (which never retains body bytes) never enters that state. On master the same scenario raised ContentLengthError spuriously, so the new behavior is arguably better, but it is an undocumented, parser-specific change hidden behind a flag named for something else (threat-model §1.12 territory).

Suggestions:

• Gate on the real condition, the parser still holding unconsumed input, independent of offloading, and add a plain-body test for it on both parsers; or restrict the guard to isinstance(self._payload, DeflateBuffer) if the intent is decompression only.
• The protocol-level data_received(b"") in connection_lost duplicates what both parsers' feed_eof already do internally (Python: self.feed_data(b"") when _payload_has_more_data; Cython: the _more_data_available loop). If it is load-bearing for some ordering, a comment saying which one would save the next reader from re-deriving it; otherwise drop it.

if self._offload_decompression and self._payload is not None and self._tail:
    return None

http_parser.py is shared by the server and the client, yet this callback casts the protocol to ResponseHandler and reads protocol._payload, then re-implements the StreamReader high-water test by poking _size, _high_water, _http_chunk_splits and _high_water_chunks. That same predicate already exists inline in three places in streams.py (feed_data at 301, end_http_chunk_receiving at 340, _read_nowait_chunk at 581), each with slightly different comparison operators, and now a fourth copy lives in a different module.

Two small moves would remove the coupling:

• Give StreamReader one predicate, for example _is_above_high_water() returning the size-or-chunk-count condition, and use it both here and in streams.py.
• Replace protocol._payload is not self.out with a protocol-level question the parser is allowed to ask, for example a BaseProtocol method payload_is_current(payload) -> bool that ResponseHandler overrides, mirroring how can_receive_data was already introduced in this PR. Then the cast("ResponseHandler", ...) and the private-attribute reads disappear.

Non-blocking; it does not change behavior, but it keeps client-only knowledge out of the shared parser.

protocol = cast("ResponseHandler", self.out._protocol)
if protocol._payload is not self.out or self.out.exception() is not None:
    self._pending = None
elif self.out._size <= self.out._high_water and (
    self.out._http_chunk_splits is None
    or len(self.out._http_chunk_splits) <= self.out._high_water_chunks
):
    protocol.resume_reading()

Risk: A cancelled executor future (e.g. executor shutdown with cancel_futures, or loop shutdown) raises asyncio.CancelledError, which is not an Exception; with the pure-Python parser it then skips HttpParser.feed_data's except Exception payload handler (http_parser.py:580) and, while the connection is still open (_deferred_eof is False), ResponseHandler.data_received re-raises it out of the future done-callback without ever calling set_exception on the StreamReader, so the consumer is left waiting on a payload whose read timeout was already dropped by pause_reading, or later receives a truncated body as a clean EOF after the forced transport close; the Cython parser's cb_on_body catches BaseException, so the two backends also diverge here.

pending, self._pending = self._pending, None
try:
    decompressed = pending.result()
except Exception as exc:
    raise ContentEncodingError(
        "Can not decode content-encoding: %s" % self.encoding
    ) from exc

Fix: Catch BaseException (or at least asyncio.CancelledError) in the pending-result branch and convert it to ContentEncodingError, or fail self.out directly, so a cancelled decode always surfaces on the payload regardless of parser backend or connection state.

Risk: This drain runs before _deferred_eof is computed, so a non-Exception error escaping data_received (the same CancelledError case as above) propagates out of connection_lost, skipping the _deferred_eof/_parser/_payload reset and super().connection_lost(), which leaves the transport attribute set and can_receive_data reporting True for a dead connection.

if (
    self._offload_decompression
    and self._parser is not None
    and not self._upgraded
):
    self.data_received(b"")

Fix: Wrap the pre-EOF drain so that any BaseException fails the current payload via _fail_deferred_payload or set_exception and then lets the rest of connection_lost's teardown proceed.