Enabling of responses compression causes blocking of the async loop

[mykola-mokhnach]

Describe the bug

Enabling response.enable_compression() on a web.Response instance causes async loop blocking delays up to 2 seconds long on our server.

I assume the original issue is coming from the fact that the compression is synchronous and eats lots of CPU cycles in case of large chunks/high core load:

aiohttp/aiohttp/http_writer.py

Line 97 in 2a1024f

chunk = self._compress.compress(chunk)

Perhaps it would be possible to "offload" the compression to another CPU core in order to unblock the main event loop.

To Reproduce

Create a simple GET endpoint, which returns a JSON response like

            response = web.Response(
                text=json.dumps(result_json, ensure_ascii=False), status=200, content_type='application/json'
            )
            response.enable_compression()
            return response

Then fetch the endpoint multiple times with json payload large enough (up to 10 megabytes in my case) and deflate compression accepted in the request headers. Async event loop debug must be enabled on the server side to see blocking events occurrences

Expected behavior

No async loop blocks should be observed longer as 100ms, although our charts show delays up to 2 seconds. This is not observed if I disable the compression.

Logs/tracebacks

[EventLoopBlockingTimeEvent] function RequestHandler._handle_request (<Handle <TaskWakeupMethWrapper object at 0x7fb12e614ee0>(<Task finishe...> result=None>)>) was blocking the loop for 168.54695370420814 ms, called with args self: <RequestHandler connected>, request: <Request GET /api/v1/files >, start_time: 2369932.481492733, request_handler: <bound method Application._handle of <Application 0x7fb12f8bd190>>

Python Version

$ python --version
Python 3.8.16

aiohttp Version

$ python -m pip show aiohttp
Version: 3.8.1

multidict Version

$ python -m pip show multidict
Version: 6.0.4

yarl Version

$ python -m pip show yarl
Version: 1.8.2

OS

python:3.8-slim-bullseye Docker image

Related component

Server

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[mykola-mokhnach]

[Dreamsorcerer]

If you can run a modified version, maybe just try wrapping it in an await asyncio.to_thread() call? If that helps, then we can consider merging it into the library.

[mykola-mokhnach]

I assume to_thread won't help there as it would still use the same core (because of GIL). Only multiprocessing could make it possible

[Dreamsorcerer]

It's the zlib module, it should be C code which releases the GIL, and should only be operating on the input, so I don't think there should be any threading issue (I could be wrong though).

[Dreamsorcerer]

The actual call being made is: https://docs.python.org/3/library/zlib.html#zlib.Compress.compress

[mykola-mokhnach]

Thanks @Dreamsorcerer
I did some experimentation and here are the results:

import asyncio
import zlib
import time
import logging
import random
import string
import concurrent.futures

MB = 1024 * 1024
logging.basicConfig(level=logging.DEBUG)


async def main(chunk: bytes) -> None:
    zlib_mode = zlib.MAX_WBITS
    compress = zlib.compressobj(wbits=zlib_mode, strategy=zlib.Z_DEFAULT_STRATEGY)
    # option 1: blocking call
    # res = compress.compress(chunk)
    # option 2: multiprocess pool
    #  with concurrent.futures.ProcessPoolExecutor() as pool:
    #    res = await asyncio.get_event_loop().run_in_executor(pool, compress.compress, chunk)
    # option 3: thread pool
    # res = await asyncio.to_thread(compress.compress, chunk)
    print(len(res))


if __name__ == '__main__':
    length = 10 * MB
    data = bytes(''.join(random.choice(string.ascii_letters) for _ in range(length)), 'utf-8')
    started_sec = time.perf_counter()
    asyncio.run(main(data), debug=True)
    print(time.perf_counter() - started_sec)

Output for option1:

DEBUG:asyncio:Using selector: KqueueSelector
7634915
0.5454047080129385
WARNING:asyncio:Executing <Task finished name='Task-1' coro=<main() done, defined at /Users/elf/code/aio_test/main.py:13> result=None created at /usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py:636> took 0.534 seconds
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>

Process finished with exit code 0

Output for option2:

DEBUG:asyncio:Using selector: KqueueSelector
WARNING:asyncio:Executing <Task finished name='Task-1' coro=<main() done, defined at /Users/elf/code/aio_test/main.py:13> exception=TypeError("cannot pickle 'zlib.Compress' object") created at /usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py:636> took 0.139 seconds
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>
concurrent.futures.process._RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/queues.py", line 244, in _feed
    obj = _ForkingPickler.dumps(obj)
  File "/usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: cannot pickle 'zlib.Compress' object
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/elf/code/aio_test/main.py", line 27, in <module>
    asyncio.run(main(data), debug=True)
  File "/usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/Users/elf/code/aio_test/main.py", line 18, in main
    res = await asyncio.get_event_loop().run_in_executor(pool, compress.compress, chunk)
  File "/usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/queues.py", line 244, in _feed
    obj = _ForkingPickler.dumps(obj)
  File "/usr/local/Cellar/[email protected]/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: cannot pickle 'zlib.Compress' object

Process finished with exit code 1

Output for option3:

DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>
7634535
0.5511908328626305

Process finished with exit code 0

So I assume we could go with Option3. I've also tried to increase the chunk size to 100MB, but still got no blocking delay on my M1 mac, which should be just fine.

[mykola-mokhnach]

I can also observe the asyncio.to_thread has been added in Python 3.9, so it probably makes sense to use a "classic" thread pool:

    res = await asyncio.get_running_loop().run_in_executor(None, compress.compress, chunk)

[Dreamsorcerer]

We can just put it in a version_info check. I wouldn't even be too bothered if we didn't thread it at all on older versions, as they won't be supported that much longer, and seems that nobody else has noticed the issue till now.

[mykola-mokhnach]

It is also important to mention another place where the sync compression is done:

aiohttp/aiohttp/web_response.py

Line 702 in 3a8e65f

async def _do_start_compression(self, coding: ContentCoding) -> None:

Setting zlib_executor_size to 0 in the Response class constructor should make the body compression non blocking:

           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)

Not sure though why the else branch still exists. I assume it should be removed for good

[Dreamsorcerer]

Appears to be done for performance:
#3201 (comment)

Ideally, we should reuse that code then, if we can get access to the same variables.

[mykola-mokhnach]

Yes, it definitely makes sense to avoid this unnecessary duplication by extracting the common code into a separate helper.
Also, we could be more advanced there, and actually measure the average sync compression time, so next time if the particular amount of data exceeds 100ms we move its compression to a separate thread if the same or larger amount of data is passed.

[Dreamsorcerer]

It's plausible, but maybe not worth the effort. If you want to create a PR for either part, then we'll look at getting it merged.

[mykola-mokhnach]

Created a PR. It's my first PR to this repository, so please let me know what needs to be done there in order to get it merged.

[bdraco]

@mykola-mokhnach If you still have performance issues with compression overhead you might try python-zlib-ng

I'm currently using it via https://github.com/bdraco/aiohttp-zlib-ng. It might be useful to integrate here in the future