Horizontal and vertical

Post from September 25, 2022, translated and technically supplemented for the redesign. “Horizontal” and “vertical” are my mental pictures, not formal categories. Original snippets are retained as historical fragments; they are not complete or currently tested.

In this post I'll try to explain my mental model of parallelism when using threads and asynchronous code. I'll also show how the approaches can be combined.

Disclaimer

This is not an academic definition of parallelism in Python. The goal is a basic understanding and avoiding some common mistakes when starting with async code.

Synchronous: vertical

Within one call, synchronous control flow progresses step by step. Loops, calls and branches of course change the path. For the sake of this mental picture I call the reading direction vertical.

# synchron code
def foo():
    print("starting here")
    a = 0
    a += 1
    b = (a + 1) / 2
    # this for loop jumps at a well-defined place.
    for x in range(10):
        b += x
    print("leaving funtion")
    return b

Top to bottom, with a jump at the loop. Nothing new, but useful for the next thought.

Threading: horizontal

Multiple threads can make progress independently. Work can truly run at the same time on multiple cores; on one core runnable threads take turns. A function generally cannot predict exactly when a switch happens. Parallel execution of Python code also depends on the interpreter and its GIL configuration. The next example illustrates my idea of horizontal execution.

Note: This and other fragments need appropriate imports; bar() must be called. Output timing and ordering are not guaranteed.

# parallel code using threading
def bar():

    # this function will be executed virtual parallel. 
    # The context switch and therefore the execution of a line of code 
    # is non-deterministic from perspective of the function.
    def foo(result):
        b = 0
        for x in range(10):
            b = b + x
            time.sleep(0.1)  # here we cannot tell that we halt here for some time. 
        result.append(b)

    results = []
    #  create 2 threads
    threads = [threading.Thread(target=foo, args=(results,)) for _ in range(2)]
    # start 2 threads "parallel"
    [t.start() for t in threads]
    # wait for them
    [t.join() for t in threads]
    print("result:", results)
    # prints result: [45, 45]

A common pattern is producer and consumer: one puts work into a queue, another takes it out. A thread-safe queue synchronizes its own operations.

# this is "sync" consumer producer pattern were the parallelism
# will be added by creating more producer oder consumer
# instead a queue one also could use a mutable buffer like a list but needs then a semaphore.
# The queue does this all for us.
def producer(q:queue.Queue):
    for work in range(100):
        time.sleep(0.001)  # extra heavy work
        q.put(work)
    q.put(None)  # poison pill
    print("producer done")


def consumer(q: queue.Queue):
    res = 0
    while True:
        item = q.get()
        if item is None:
            break
        else:
            res = res + item
    print("consumer result", res)


def main():
    q = queue.Queue()
    threads = [threading.Thread(target=target, args=(q,)) for target in [producer, consumer]]
    [t.start() for t in threads]
    [t.join() for t in threads]
    print("bye")

With multiple consumers, a single termination signal is not automatically enough for all of them. Threads and context switches have costs; whether they matter depends on the application. Concurrency also has its pitfalls.

Asynchronous: vertical

With asyncio, coroutines can hand over control at suitable await points, allowing the event loop to run other work. A single coroutine still reads top to bottom. That's why I call it vertical in this model.

# a simple asynchron function will execute vertical line by line 
# allowing at defined places to halt(wait) and resume for
# other code to be executed
async def foo():
    print("starting here")
    a = 1
    await asyncio.sleep(1)  # at this place other code can be executed
    print("slept for 1 second")
    a += 2
    await asyncio.sleep(0.5)  # at this place other code can be executed
    print("leaving")
    return a

For comparison, here's the original example with time.sleep. Correction: It blocks the event loop's thread. Releasing the GIL during time.sleep may allow other threads to progress, but not other tasks on the same blocked event loop. An async def does not make a blocking call non-blocking.

async def foo():
    print("starting here")
    a = 1
    time.sleep(1)  # at this place other code can be executed
    print("slept for 1 second")
    a += 2
    time.sleep(0.5)  # at this place other code can be executed
    print("leaving")
    return a

Combining both

An event loop can manage several tasks: each request coroutine can wait for I/O while others progress. This is concurrency; it doesn't automatically mean parallel CPU execution of Python code on a single event-loop thread.

Keep references to background tasks: the loop keeps weak references, so a task can potentially be collected before it finishes. See the Python documentation.

async def request(page:int,  client: aiohttp.ClientSession):
        response = await client.get(f"https://books.toscrape.com/catalogue/page-{page}.html")
        if response.status < 300:
            body = await response.text()
            print(body[:18].strip())


async def main():
    client = aiohttp.ClientSession()
    loop = asyncio.get_event_loop()
    time_taken = time.time()
    async with client:
        # hold reference here
        pending = [loop.create_task(request(page, client)) for page in range(10)]
        # wait for them and release when all complete (default of wait())
        done, pending = await asyncio.wait(pending, return_when=asyncio.ALL_COMPLETED)
    print(f"done in {time.time() - time_taken:.2f}")
    await client.close()
    await asyncio.sleep(0.5)

if __name__ == '__main__':
    asyncio.run(main(), debug=True)
    # prints 
    # <!DOCTYPE html>
    # ...
    # done in 1.02

The example starts ten requests nearly together. The displayed times come from the original run and are not benchmarks. This fragment requires imports and uses historical APIs; the target site and its page numbering can change.

In the counterexample below, the list comprehension awaits each request before starting the next. The function is asynchronous, but these requests run sequentially:

async def request(page: int, client: aiohttp.ClientSession):
    response = await client.get(f"https://books.toscrape.com/catalogue/page-{page}.html")
    if response.status < 300:
        body = await response.text()
        print(body[:18].strip())


async def main():
    client = aiohttp.ClientSession()
    loop = asyncio.get_event_loop()
    time_taken = time.time()
    async with client:
        results = [await request(page, client) for page in range(10)]
    print(f"done in {time.time() - time_taken:.2f}")
    await client.close()
    await asyncio.sleep(0.5)


if __name__ == '__main__':
    asyncio.run(main(), debug=True)
    # prints
    # <!DOCTYPE html>
    # ...
    # done in 2.50

In the end: read the docs, make small experiments and measure performance rather than guessing. More on asyncio and processes is in the follow-up post.

Horizontal and vertical.

Disclaimer

Synchronous: vertical

Threading: horizontal

Asynchronous: vertical

Combining both