Sync serializer/deserializer calls block the async event loop

[lukasjesche]

Hello!

The serialize() and deserialize() methods on the Worker are synchronous and called directly from async methods without being offloaded to a thread. For non-trivial payloads or serializers, this can blocks the event loop and can stall the entire worker.

This could either be fixed by offloading the sync serializer/deserializer calls to a worker thread via anyio.to_thread.run_sync() or allow the Worker to accept async serializer/deserializer callables instead that can run async agnostic.

Environment:
streaq version: 6.2.1 (also affects earlier versions)
Reviewed from master branch, streaq/worker.py

[Graeme22]

Offloading to a thread would make more sense since serialization/deserialization is always going to be CPU bound.

I'm curious to see how much overhead threading has in the average case, I guess hopefully not too much?

Would you like to open a PR? Make sure you're familiar with the anyio docs on threading: https://anyio.readthedocs.io/en/stable/threads.html

[Graeme22]

Of course, this will only benefit in free-threaded mode so we'd want to detect that and only use threads in that case

[lukasjesche]

While free-threaded Python will be great for true parallel execution, I don't think the threading should be limited to that mode.
The critical benefit is that offloading to a thread turns a hard block into an await point, giving the event loop a chance to make progress on other coroutines while (de)serialization happens. In most cases this will allow the service to react and process as normal. Without it, the loop is completely frozen until the call returns.

The worker already offloads sync tasks via asyncify regardless of free-threading, and I think serialization should follow that example.
As for overhead, it'll depend on payload size and exact serializer (and Python version) so hard to say.

[Graeme22]

While free-threaded Python will be great for true parallel execution, I don't think the threading should be limited to that mode. The critical benefit is that offloading to a thread turns a hard block into an await point, giving the event loop a chance to make progress on other coroutines while (de)serialization happens. In most cases this will allow the service to react and process as normal. Without it, the loop is completely frozen until the call returns.

There are already yield points before and after deserialization in the run_task function, so I don't see how having another yield point changes anything. Since this is CPU-bound work, there will not be any benefit if the GIL is enabled (in fact, performance might be slightly worse due to threading overhead)--you can try yourself to verify.

The worker already offloads sync tasks via asyncify regardless of free-threading, and I think serialization should follow that example. As for overhead, it'll depend on payload size and exact serializer (and Python version) so hard to say.

The difference is that sync tasks are expected to perform I/O or make syscalls (which release the GIL), whereas serialization/deserialization is purely CPU-bound work that prevents real parallelism.

[lukasjesche]

I'm not sure I follow the argument that serialization/deserialization is not gonna release the GIL.

I'm more familar with asyncio so here is an example I have with that:

@pytest.mark.asyncio
async def test_async():
    import time
    import zlib
    from collections import defaultdict
    import asyncio
    import json

    alive_counter = 0

    async def heartbeat():
        while True:
            print("alive")
            nonlocal alive_counter
            alive_counter += 1
                        await asyncio.sleep(0.1)

    d = defaultdict(lambda: defaultdict(list))
    for x in range(1000):
        for y in range(1000):
            d[x][y].append(x * y)

    task = asyncio.create_task(heartbeat())

    def blocker(d):
        compressed = zlib.compress(json.dumps(d).encode('utf-8'))
        decompressed = zlib.decompress(compressed)

    start = time.time()
    blocker(d)
    print(f"sync execution time {time.time() - start}")
    print(f"sync execution alive counter: {alive_counter}")
    assert alive_counter == 0

    start = time.time()
    await asyncio.to_thread(blocker, d)
    print(f"async execution time {time.time() - start}")
    print(f"async execution alive counter: {alive_counter}")
    assert alive_counter > 0

    task.cancel()

output:

sync execution time 1.2167742252349854
sync execution alive counter: 0
alive
alive
alive
alive
alive
alive
alive
alive
alive
alive
async execution time 1.1736228466033936
async execution alive counter: 10

Based on this my assumption would be that potential overhead would be still worth the non-blocking eventloop, even if there will be cases where the GIL is blocked.

[Graeme22]

I'm not sure I follow the argument that serialization/deserialization is not gonna release the GIL.

I'm more familar with asyncio so here is an example I have with that:

I'm not sure what this is supposed to demonstrate, the conversation is about threading no?

[lukasjesche]

My main concern is about blocking code that can stop the worker from being responsive.

The example I provided showcases that a big chunk of data that is dumped and compressed is able to block any async task running in the event loop stopping all async tasks in the background.
When run within a seperate threadpool the async loop can continue normally and the task can run regularly as it should.

With my understanding this means depending on what the serializer does the whole worker stops doing anything and becomes unresponsive. This can cause connection losses or even cause restart if the worker since it might seem unhealthy.

[Graeme22]

With my understanding this means depending on what the serializer does the whole worker stops doing anything and becomes unresponsive. This can cause connection losses or even cause restart if the worker since it might seem unhealthy.

Your understanding is correct. I'm just wanting to work through exactly how threading will help.

[lukasjesche]

Okay, thats why I posted the example code.
Adding the sync work to a new thread left the background task (and event loop) continue working. Even without free-threaded python running.
While this will not always work there due to the GIL it can help for cases like I shown in the example code (using json.loads/dumps, zlib de/compression). So I would think this is worth the potential trade-off with any overhead for the threading.

[Graeme22]

Okay, thats why I posted the example code.
Adding the sync work to a new thread left the background task (and event loop) continue working. Even without free-threaded python running.
While this will not always work there due to the GIL it can help for cases like I shown in the example code (using json.loads/dumps, zlib de/compression). So I would think this is worth the potential trade-off with any overhead for the threading.

I think the example only shows improvement because zlib is being used, and zlib releases the GIL during computation. I believe json does not, nor similar serialization libraries.

I would accept a PR allowing async serialization functions, but I definitely don't want to add threading for little to no benefit for most users.