Asyncify string joining in Python

More in depth explanation about my comment:

Asyncio is a great tool if your processor has a lot of waiting to do. For example: when you make request to a db over the network, after the request is sent your cpu just does nothing until it gets an answer.

Using the async await syntax you can have your processor execute other tasks while "waiting" for the current one to finish. this does not mean it runs them in parallel. There is only one task running at a time.

In your case (for what i can see) the cpu never waits for something it is constantly running string operations.

if you want to run these operations in parallel you might want to take a look at ProcesPools. This is not bound by a single process and core but will spread the processing over several cores to run it in parallel.

from concurrent.futures import ProcessPoolExecutor

def main():
    with ProcessPoolExecutor() as executor:
        transformed_data = executor.map(do_some_transformation, data) #returns an iterable

    stacked_jsons = "\n\n".join(json.dumps(t, separators=(",", ":")) for t in transformed_data)

if __name__ == '__main__':
    main()

I hope the provided code can help you.

ps. The if __name__ part is required

edit: i saw your comment about 10k dicts, assume you have 8 cores (ignore multithreading) then each process will only transform 1250 dicts, instead of the 10k your main thread does now. These processes run simultaniously and although the performance increase is not linear it should process them a lot faster.

The point of str.join is to transform an entire list at once.¹ If items arrive incrementally, it can be advantageous to accumulate them one by one.

async def join(by: str, _items: 'AsyncIterable[str]') -> str:
    """Asynchronously joins items with some string"""
    result = ""
    async for item in _items:
        if result and by:  # only add the separator between items
            result += by
        result += item
    return result

The async for loop is sufficient to let the async iterable suspend between items so that other tasks may run. The primary advantage of this approach is that even for very many items, this never stalls the event loop for longer than adding the next item.

This utility can directly digest the async generator:

stacked_jsons = join("\n\n", (json.dumps(event, separators=(",", ":")) async for t in transformed_data))

When it is know that the data is small enough that str.join runs in adequate time, one can directly convert the data to a list instead and use str.join:

stacked_jsons = "\n\n".join([json.dumps(event, separators=(",", ":")) async for t in transformed_data])

The [... async for ...] construct is an asynchronous list comprehension. This internally works asynchronously to iterate, but produces a regular list once all items are fetched – only this resulting list is passed to str.join and can be processed synchronously.

¹ Even when joining an iterable, str.join will internally turn it into a list first.

TL;DR: Consider using producer/consumer pattern, if do_some_transformation is IO bound, and you really want an incremental aggregation.

Of course, async itself only brings an advantage if you actually have any other proper async tasks to begin with.

As @MisterMiyagi said, if do_some_transformation is IO bound and time consuming, firing all transformation as a horde of async tasks can be a good idea.

Example code:

import asyncio
import json


data = ({"large": "data"},) * 3  # large
stacked_jsons = ""


async def transform(d: dict, q: asyncio.Queue) -> None:
    # `do_some_transformation`: long IO bound task
    await asyncio.sleep(1)
    await q.put(d)


# WARNING: incremental concatination of string would be slow,
# since string is immutable.
async def join(q: asyncio.Queue):
    global stacked_jsons
    while True:
        d = await q.get()
        stacked_jsons += json.dumps(d, separators=(",", ":")) + "\n\n"
        q.task_done()


async def main():
    q = asyncio.Queue()
    producers = [asyncio.create_task(transform(d, q)) for d in data]
    consumer = asyncio.create_task(join(q))
    await asyncio.gather(*producers)
    await q.join()  # Implicitly awaits consumers, too
    consumer.cancel()
    print(stacked_jsons)


if __name__ == "__main__":
    import time

    s = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

So that do_some_transformation don't block each other. Output:

$ python test.py
{"large":"data"}
{"large":"data"}
{"large":"data"}

test.py executed in 1.00 seconds.

Besides, I don't think incremental concatenation of string is a good idea, since string is immutable and a lot of memory would be wasted ;)

Reference: Async IO in Python: A Complete Walkthrough - Real Python