Asyncio support without threading

[sashkent3]

Under the hood async_stream_zip uses threads as a layer over the synchronous stream_zip function. Would you accept a more flexible 'pure-asyncio' implementation? This would still allow running async_stream_zip via threads, however in my limited testing simply using asyncio.run provides a 2-5% performance benefit over the current async_stream_zip implementation.

I have thrown together a quick and dirty implementation which is intentionally kept as close to the original as possible. I'm happy to contribute it as a PR for further review and refinement.

[michalc]

Would you accept a more flexible 'pure-asyncio' implementation?

Hmmm... I guess maybe. I'm torn about the duplication that would then have to be maintained indefinitely, kinda doubling the work for all changes in a way. I would love a pure asyncio implementation that doesn't have duplication with the sync implementation...

however in my limited testing simply using asyncio.run provides a 2-5% performance benefit over the current async_stream_zip

Can you give more details on this? Specifically since asyncio.run runs a coroutine, so to me it's not obvious how this can replace the coroutine async_stream_zip

[sashkent3]

I would love a pure asyncio implementation that doesn't have duplication with the sync implementation...

I understand your concerns, however, avoiding the code duplication while maintaining proper implementations for both cases isn't possible to my knowledge. The best alternative I've considered is maintaining the async implementation and rewriting the sync implementation through it. For example:

import asyncio


async def async_stream_zip():
    # implementation goes here
    pass


def sync_stream_zip():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    return loop.run_until_complete(async_stream_zip())

I'm not 100% certain though if such an implementation is fully backwards compatible (in the context of running sync stream_zip from an async environment). The most common practice is to maintain split implementations.

Can you give more details on this?

This is the crucial line in the current implementation. It essentially runs stream-zip in a thread pool. In contrast, asyncio.run uses its own scheduler based on non-blocking IO. This is often more efficient because it eliminates the need to spawn relatively heavy OS threads (see Green thread on Wikipedia). Thurthermore, asyncio.run would not be included in the implementation at all. We should simply provide the async definition and leave the method of running it up to the user (asyncio, uvloop, greenlet, eventlet, etc.).

[michalc]

The best alternative I've considered is maintaining the async implementation and rewriting the sync implementation through it. For example: [...] I'm not 100% certain though if such an implementation is fully backwards compatible (in the context of running sync stream_zip from an async environment)

My suspicion is that it indeed won't be fully backwards compatible in a number of ways, and to have an event loop running in what is otherwise regular Python sync code makes it seem somehow... over-engineered... And maybe even introduce a performance penalty of its own

Thurthermore, asyncio.run would not be included in the implementation at all.

Ah yes I see

The most common practice is to maintain split implementations.

So we would do this I suspect. However, for a 2-5% performance benefit... I think I'm a bit anti. I can't help but suspect the current asyncio performance is good enough for almost everyone (maybe even including... you?) and squeezing a few % at the price of an increase risk of incorrect behaviour or slower time to address issues in future (by myself or other future maintainers) because of maintaining almost-identical implementations, might not be worth it.

[michalc]

So there is another option that's crossing my mind... it is possible factor out all the logic somehow to some core code that is used by both the sync and async versions? Not sure what that would look like, but it should be possible?

[michalc]

It could maybe define a StreamZipper or something, working like this a bit. For sync:

def stream_zip(member_files):
    start_file, flush_zip = StreamZipper()

    for member_file in member_files:
        push_data, flush_member_file = start_file(member_file[0:4])

        for chunk in member_file[4]:
            yield push_data(chunk)

        yield flush_member_file()

    yield flush_zip()

and then for async:

async def async_stream_zip(member_files):
    start_file, flush_zip = StreamZipper()

    async for member_file in member_files:
        push_data, flush_member_file = start_file(member_file[0:4])

        async for chunk in member_file[4]:
            yield push_data(chunk)

        yield flush_member_file()

    yield flush_zip()

Might not have thought it all through, but that's the sort of idea...

[Diegorro98]

Hi! Are you planning to develop this?

I have an use case where I would like to send the chunks of the zip file while the files that compound it are being generated through an streaming response in FastAPI.
This files are generated and then yield asynchronously, and each of them can take some time to be generated, so in the meantime I would like to send the chunks from the ones that have been already generated.

[michalc]

@Diegorro98

Are you planning to develop this?

I think probably not in the near future, but it's not definitely not a "no we don't want this"

But also, I think would like to know more on the real-world consequence of the current threads-based implementation. How bad, by some measure, is the current implementation / how good can we make it? In your specific case as well in FastAPI, is the current setup not good enough, and if no, why not?

[Diegorro98]

The actual implementation waits for all files to be generated before start sending the chunks of data.
If there are many files that can take a bit to generate, like in our case, first byte can take pretty long to be sent, which with some clients can lead to timeout. We can (and we already do) increase the timeout, but because it is not known how long by the client how long can it take to generate the whole zip, sending chunks of bytes from the files while they are being generated could help us.

An example:
Having this FastAPI code:

import asyncio
from datetime import datetime
from stat import S_IFREG

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from stream_zip import ZIP_32, async_stream_zip

app = FastAPI()


async def test():
    await asyncio.sleep(4)  # Simulate some processing delay
    yield b"First chunk"
    await asyncio.sleep(4)  # Simulate some processing delay
    yield b"Second chunk"


@app.get("/streaming-chunks")
async def zip_pro():
    return StreamingResponse(
        test(),
    )


async def async_member_files():
    async def file_content(data: bytes):
        yield data

    await asyncio.sleep(4)  # Simulate some processing delay
    yield (
        "my-file-1.txt",
        datetime.now(),
        S_IFREG | 0o600,
        ZIP_32,
        file_content(b"first file contents"),
    )
    await asyncio.sleep(4)  # Simulate some processing delay
    yield (
        "my-file-2.txt",
        datetime.now(),
        S_IFREG | 0o600,
        ZIP_32,
        file_content(b"second file contents"),
    )


@app.get("/zip")
async def zip():
    return StreamingResponse(
        async_stream_zip(async_member_files()),
        media_type="application/zip",
        headers={"Content-Disposition": 'attachment; filename="stream.zip"'},
    )

and this python script as a client that fetches those endpoints:

import asyncio
import datetime

from httpx import AsyncClient
import httpx


async def download_dataset():
    async with AsyncClient(base_url="http://127.0.0.1:8080") as ac:
        start_time = datetime.datetime.now()
        chunks = 0

        try:
            async with ac.stream("GET", "/streaming-chunks", timeout=10) as response:
                print("Starting download from '/streaming-chunks'...")
                async for _ in response.aiter_raw():
                    chunks += 1
                    print(f"Received a chunk number {chunks} at {datetime.datetime.now() - start_time}")
            end_time = datetime.datetime.now()
            elapsed = end_time - start_time
            print(
                f"Download from '/streaming-chunks' completed in {elapsed.total_seconds()} seconds."
            )
        except httpx.ReadTimeout:
            print(f"Timeout error occurred, chunks received: {chunks}")

        chunks = 0
        try:
            async with ac.stream("GET", "/zip", timeout=5) as response:
                print("Starting download from '/zip'...")
                async for _ in response.aiter_raw():
                    chunks += 1
                    print(f"Received a chunk number {chunks} at {datetime.datetime.now() - start_time}")
            end_time = datetime.datetime.now()
            elapsed = end_time - start_time
            print(
                f"Download from '/zip' completed in {elapsed.total_seconds()} seconds."
            )
        except httpx.ReadTimeout:
            print(
                f"Timeout error occurred, chunks received from '/zip' so far: {chunks}"
            )


if __name__ == "__main__":
    asyncio.run(download_dataset())

The output will be:

Starting download from '/streaming-chunks'...
Received a chunk number 1 at 0:00:04.020238
Received a chunk number 2 at 0:00:08.024703
Download from '/streaming-chunks' completed in 8.025434 seconds.
Starting download from '/zip'...
Timeout error occurred, chunks received from '/zip' so far: 0

[michalc]

The actual implementation waits for all files to be generated before start sending the chunks of data.

Ah! So right now I think this is pretty independent of the fact the asyncio interface is powered by threads, and in fact if we were to make it pure asyncio, the behaviour would be the same...

... because (barring a bug), it doesn't wait for all files to be generated, and actually stream-zip waits for chunk_size bytes to be ready to send, which by default is 64k, which in the case above, never happens and so the bytes don't get sent until the after all the sleeps.

You could reduce the chunk_size parameter, see how you get on, but...

first byte can take pretty long to be sent

I guess it crosses my mind that timeouts could still happen, because there would still be periods of time when no bytes would be sent. It feels like the data source is basically non-streaming, which means that trying to put a streaming interface in front of it is always(?) going to be a bit brittle / risk timeouts?

Maybe there is some other behaviour stream-zip could do though... not sure have to admit

[michalc]

Maybe there is some other behaviour stream-zip could do though

So it could yield the local header of each member file as soon as its generated, independent of chunk_size, but I guess I do suspect that client code might still hit a timeout. I think in a lot of cases the client timeout isn't based on "time to first byte" but more "time when not sending any bytes"?

(Also, I realise this is pretty far from the original topic of this issue, so if you would like to continue, I think raising another issue would be best)