body thread helpers

CHANGES/3205.feature

@@ -0,0 +1 @@
+Add zlib_executor_size Response parameter to allow compression to run in a background executor to avoid blocking the main thread and potentially triggering health check failures.

aiohttp/web_response.py

@@ -7,6 +7,7 @@
 import time
 import warnings
 import zlib
+from concurrent.futures import Executor
 from email.utils import parsedate
 from http.cookies import SimpleCookie
 from typing import (TYPE_CHECKING, Any, Dict, Iterator, Mapping,  # noqa
@@ -310,7 +311,7 @@ def _generate_content_type_header(
             ctype = self._content_type
         self._headers[CONTENT_TYPE] = ctype
 
-    def _do_start_compression(self, coding: ContentCoding) -> None:
+    async def _do_start_compression(self, coding: ContentCoding) -> None:
         if coding != ContentCoding.identity:
             assert self._payload_writer is not None
             self._headers[hdrs.CONTENT_ENCODING] = coding.value
@@ -319,15 +320,15 @@ def _do_start_compression(self, coding: ContentCoding) -> None:
             # remove the header
             self._headers.popall(hdrs.CONTENT_LENGTH, None)
 
-    def _start_compression(self, request: 'BaseRequest') -> None:
+    async def _start_compression(self, request: 'BaseRequest') -> None:
         if self._compression_force:
-            self._do_start_compression(self._compression_force)
+            await self._do_start_compression(self._compression_force)
         else:
             accept_encoding = request.headers.get(
                 hdrs.ACCEPT_ENCODING, '').lower()
             for coding in ContentCoding:
                 if coding.value in accept_encoding:
-                    self._do_start_compression(coding)
+                    await self._do_start_compression(coding)
                     return
 
     async def prepare(
@@ -359,7 +360,7 @@ async def _start(self, request: 'BaseRequest') -> AbstractStreamWriter:
             headers.add(hdrs.SET_COOKIE, value)
 
         if self._compression:
-            self._start_compression(request)
+            await self._start_compression(request)
 
         if self._chunked:
             if version != HttpVersion11:
@@ -479,7 +480,9 @@ def __init__(self, *,
                  text: Optional[str]=None,
                  headers: Optional[LooseHeaders]=None,
                  content_type: Optional[str]=None,
-                 charset: Optional[str]=None) -> None:
+                 charset: Optional[str]=None,
+                 zlib_executor_size: Optional[int]=None,
+                 zlib_executor: Executor=None) -> None:
         if body is not None and text is not None:
             raise ValueError("body and text are not allowed together")
 
@@ -533,6 +536,8 @@ def __init__(self, *,
             self.body = body
 
         self._compressed_body = None  # type: Optional[bytes]
+        self._zlib_executor_size = zlib_executor_size
+        self._zlib_executor = zlib_executor
 
     @property
     def body(self) -> Optional[Union[bytes, Payload]]:
@@ -652,19 +657,34 @@ async def _start(self, request: 'BaseRequest') -> AbstractStreamWriter:
 
         return await super()._start(request)
 
-    def _do_start_compression(self, coding: ContentCoding) -> None:
+    def _compress_body(self, zlib_mode: int) -> None:
+        compressobj = zlib.compressobj(wbits=zlib_mode)
+        body_in = self._body
+        assert body_in is not None
+        self._compressed_body = \
+            compressobj.compress(body_in) + compressobj.flush()
+
+    async def _do_start_compression(self, coding: ContentCoding) -> None:
         if self._body_payload or self._chunked:
-            return super()._do_start_compression(coding)
+            return await super()._do_start_compression(coding)
+
         if coding != ContentCoding.identity:
             # Instead of using _payload_writer.enable_compression,
             # compress the whole body
             zlib_mode = (16 + zlib.MAX_WBITS
                          if coding == ContentCoding.gzip else -zlib.MAX_WBITS)
-            compressobj = zlib.compressobj(wbits=zlib_mode)
             body_in = self._body
             assert body_in is not None
-            body_out = compressobj.compress(body_in) + compressobj.flush()
-            self._compressed_body = body_out
+            if self._zlib_executor_size is not None and \
+                    len(body_in) > self._zlib_executor_size:
+                await asyncio.get_event_loop().run_in_executor(
+                    self._zlib_executor, self._compress_body, zlib_mode)
+            else:
+                self._compress_body(zlib_mode)
+
+            body_out = self._compressed_body
+            assert body_out is not None
+
             self._headers[hdrs.CONTENT_ENCODING] = coding.value
             self._headers[hdrs.CONTENT_LENGTH] = str(len(body_out))
 

docs/web_reference.rst

@@ -805,7 +805,8 @@ Response
 ^^^^^^^^
 
 .. class:: Response(*, body=None, status=200, reason=None, text=None, \
-   headers=None, content_type=None, charset=None)
+   headers=None, content_type=None, charset=None, zlib_executor_size=sentinel,
+   zlib_executor=None)
 
    The most usable response class, inherited from :class:`StreamResponse`.
 
@@ -830,6 +831,15 @@ Response
    :param str charset: response's charset. ``'utf-8'`` if *text* is
                        passed also, ``None`` otherwise.
 
+   :param int zlib_executor_size: length in bytes which will trigger zlib compression
+                            of body to happen in an executor
+
+      .. versionadded:: 3.5
+
+   :param int zlib_executor: executor to use for zlib compression
+
+      .. versionadded:: 3.5
+
 
    .. attribute:: body
 

tests/test_web_response.py

@@ -1,7 +1,9 @@
 import collections
 import datetime
+import gzip
 import json
 import re
+from concurrent.futures import ThreadPoolExecutor
 from unittest import mock
 
 import pytest
@@ -418,6 +420,32 @@ async def test_force_compression_no_accept_gzip() -> None:
     assert 'gzip' == resp.headers.get(hdrs.CONTENT_ENCODING)
 
 
+async def test_change_content_threaded_compression_enabled() -> None:
+    req = make_request('GET', '/')
+    body_thread_size = 1024
+    body = b'answer' * body_thread_size
+    resp = Response(body=body,
+                    zlib_executor_size=body_thread_size)
+    resp.enable_compression(ContentCoding.gzip)
+
+    await resp.prepare(req)
+    assert gzip.decompress(resp._compressed_body) == body
+
+
+async def test_change_content_threaded_compression_enabled_explicit() -> None:
+    req = make_request('GET', '/')
+    body_thread_size = 1024
+    body = b'answer' * body_thread_size
+    with ThreadPoolExecutor(1) as executor:
+        resp = Response(body=body,
+                        zlib_executor_size=body_thread_size,
+                        zlib_executor=executor)
+        resp.enable_compression(ContentCoding.gzip)
+
+        await resp.prepare(req)
+        assert gzip.decompress(resp._compressed_body) == body
+
+
 async def test_change_content_length_if_compression_enabled() -> None:
     req = make_request('GET', '/')
     resp = Response(body=b'answer')