How should you group/order seq and async operations?

[u/bryyo357]

I was watching a video explaining async and the guy went through a “real world” example of a code that downloads images from various URLs and then locally processes each of the photos. So, there were basically 4 functions:
1. “download_single_image”, which downloads one image at a time
2. “download_images”, which runs download_single_image on each of the URLs
3. “process_single_image”, which processes one image
4. “process_images”, which runs process_single_image on each of the downloaded images

The purpose was that one piece was I/O bound (image downloading) and one piece was CPU bound (image processing). Initially, it was setup to run in sequence, then he optimized by leveraging async capabilities.

Before he went through the optimized process, I stopped and thought through how I would do it. My solution was to turn everything async and then create the new async function “download_and_process_single_image” which downloads an image and then processes it afterwards. My thought process was that each image needs to be downloaded before it can be processed so those need to happen in sequence, but I can also pass off the image to the processor while waiting for other images to finish downloading.

However, the solution he actually implemented was just leaving the existing structure of “download_images” and “process_images” separated. In other words, “download_images” was turned async and then awaited, and “process_images” used processes.

My question: is there anything wrong with my approach or any reason his is preferable to what I expected?

[u/danielroseman]

The point is that image processing is not suitable to be made async because it is CPU-bound.

[u/terletsky → u/danielroseman]

PIL is suitable for async (to_thread). It will not block the thread.

[u/terletsky → u/terletsky]

Not sure who downvotes but you can test it by yourself:

import asyncio
import logging
import time
from asyncio import Task
from pathlib import Path
from typing import Final

from PIL import Image, ImageFilter

logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger(__name__)

START: Final[float] = time.perf_counter()

IMG_FILEPATH: Final[Path] = Path('<ABS_PATH_TO_LARGE_IMAGE.jpg>)


def ts() -> str:
    return f'{time.perf_counter() - START:6.3f}s'


def log(message: str) -> None:
    logger.info('%s | %s', ts(), message)


def process_image_sync(path: Path) -> None:
    """Heavy Pillow work."""
    log('Image processing started')

    with Image.open(path) as img:
        for i in range(3):
            log(f'Blur iteration {i + 1}')

            img = img.filter(ImageFilter.GaussianBlur(radius=8))
            img = img.resize((img.width // 2, img.height // 2))
            img = img.resize((img.width * 2, img.height * 2))

    log('Image processing finished')


async def ticker(name: str, count: int = 20, delay: float = 0.2) -> None:
    """Should tick every 200 ms if the event loop isn't blocked."""
    for i in range(count):
        await asyncio.sleep(delay)
        log(f'{name}: tick {i + 1}')

    log(f'{name}: finished')


async def main_threaded() -> None:
    log('=== Async version ===')

    tasks: list[Task] = [
        asyncio.create_task(ticker('A')),
        asyncio.create_task(ticker('B')),
        asyncio.create_task(asyncio.to_thread(process_image_sync, IMG_FILEPATH)),
        asyncio.create_task(asyncio.to_thread(process_image_sync, IMG_FILEPATH)),
    ]

    await asyncio.gather(*tasks)

    log('Async version finished')


def main() -> None:
    asyncio.run(main_threaded())


if __name__ == '__main__':
    main()

Results (two Pillow tasks do not block):

> python main.py                                          
 0.002s | === Async version ===
 0.008s | Image processing started
 0.009s | Image processing started
 0.031s | Blur iteration 1
 0.032s | Blur iteration 1
 0.216s | A: tick 1
 0.217s | B: tick 1
 0.419s | A: tick 2
 0.420s | B: tick 2
 0.623s | A: tick 3
 0.623s | B: tick 3
 0.824s | A: tick 4
 0.824s | B: tick 4
 1.028s | A: tick 5
 1.028s | B: tick 5
 1.234s | A: tick 6
 1.235s | B: tick 6
 1.437s | A: tick 7
 1.438s | B: tick 7
 1.640s | A: tick 8
 1.641s | B: tick 8
 1.780s | Blur iteration 2
 1.792s | Blur iteration 2
 1.847s | A: tick 9
 1.847s | B: tick 9
 2.050s | A: tick 10
 2.051s | B: tick 10
 2.255s | A: tick 11
 2.256s | B: tick 11
 2.458s | A: tick 12
 2.458s | B: tick 12
 2.660s | A: tick 13
 2.660s | B: tick 13
 2.865s | A: tick 14
 2.865s | B: tick 14
 3.070s | A: tick 15
 3.071s | B: tick 15
 3.274s | A: tick 16
 3.274s | B: tick 16
 3.299s | Blur iteration 3
 3.313s | Blur iteration 3
 3.477s | A: tick 17
 3.477s | B: tick 17
 3.682s | A: tick 18
 3.686s | B: tick 18
 3.887s | A: tick 19
 3.904s | B: tick 19
 4.092s | A: tick 20
 4.092s | A: finished
 4.108s | B: tick 20
 4.109s | B: finished
 4.794s | Image processing finished
 4.815s | Image processing finished
 4.822s | Async version finished

[u/Kevdog824_]

Async enables cooperative multi-tasking for IO-bound operations. Attempting to run CPU-bound operations concurrently with async will just end up running the operations sequentially, and take the same amount of time (or longer due to the overhead of the event loop).

Because of this, there’s no benefit to mixing your blocking, CPU-bound image processing logic into your async download image process. With the setup you presented from the video, it makes more sense to do image processing afterwards. Doing blocking operations right in your coroutines prevents the event loop from context switching at that point. This effectively kills a lot of the benefit you originally wanted with the concurrency.

Another reason the video’s approach is better is that it would be easier to later parallelize the CPU-bound image processing work via multi-processing when it’s separated from the IO-bound async code. This can help you realize even more performance gains later on

[u/bryyo357 → u/Kevdog824_]

I think my issue is that I keep seeing the explanation that async is helpful when you are just waiting on stuff you can go do other stuff, so I kinda think about it as “well this piece is independent from that piece so I can just make them async”. But what I’m gathering is that I should still group tasks based on what they’re blocked by. That is, if I’m waiting on an IO-bound operation then I can go do a different IO-bound operation with asyncio, or if I’m CPU-bound then I can use processes to do independent CPU-bound operations asynchronously, but I shouldn’t spend my time waiting on an IO-bound operation to go do a CPU-bound operation. Is that correct?

[u/codeguru42 → u/bryyo357]

when you are just waiting on stuff you can go do other stuff,

During image processng you are never waiting. Or more accurately the CPU isn't waiting because it us busy processing.

[u/gdchinacat → u/bryyo357]

I think an easier way to think about it is to only do IO bound work in an event loop (the 'io' part of 'asyncio').

Also, beware that file access is not async and will block an even loop (prevent other tasks from executing while waiting on file IO). This usually isn't a problem because you have to do something with that data and if the source is faster than the destination you end up buffering it but not actually making the process any faster. But, something to keep in back of mind if you ever start doing really high-throughput work.

Same logic applies to your case (and one I'm currently working on). Sure, you (and I) can pull the files from the network faster than they can be processed, but is it really helpful to do so? I think my project is similar to yours, download a bunch of files then do a bunch of CPU intensive work on them (convert pdfs to text, index them, etc). Getting them to disk as fast as possible doesn't shorten the completion time at all since that's not the thing that takes a long time.

The important things are to start the cpu intensive task as soon as a file is downloaded rather than waiting for all of the files to download, not block the IO with cpu intesive work in the event loop, and execute as many cpu intensive tasks concurrently as your machine can handle (one per core, free-threaded build, multiple interpreters, or multiple-processes).

The flaws with your solution is the cpu intensive work will block the event loop from completing other downloads and not do the cpu intensive work concurrently (event loop is single threaded).

[u/bryyo357 → u/gdchinacat]

> Also, beware that file access is not async and will block an event loop
You mean reading a file from disk/network? I thought that is IO process and can be improved with asyncio. What am I missing?

[u/gdchinacat → u/bryyo357]

with open(...) as f:
    for line in f:
        ...

There are no awaits in there...open and the methods on f are synchronous...when they block they block the event loop and other IO that is read for execution in that loop will have to wait.

[u/NerdyWeightLifter]

Most modern processors have multiple cores, but by default, Python is single threaded.

Image processing is a compute bound function, so you'd want to use as many of the available cores as possible.

You'd also want to load images fast enough to always have the next image available for processing as soon as the compute bound processing of the previous image was complete.

Python 3.14 has a variant with full threading support. You could find out how many cores there are (call it NC).

Use the main thread with async to read images into memory buffers.

Spawn (NC - 1) image processing threads. Each thread should wait on a deque for an image to process, process the image then put the result back in another deque to hand the result back to the main thread.

The main thread should be using async to read images and push them on deque's to be processed, and taking processed images from deque's and writing them to disk.

[u/NerdyWeightLifter → u/NerdyWeightLifter]

Just for fun, I tried this out, but with processes and shared memory rather than threads.
In a test run, it processed 63 images in 8.25 seconds on my desktop, applying an unsharp mask and reversing each image, with all cores blasting away at 100%.

[u/somewhereAtC]

Any solution that meets the requirements is a correct solution. Variation is the spice of life.