Async playground

README.md

Playground for async code

Playground with async code visualization. Each line is separate coroutine.

Legend

Each char in outpput represents 0.1 second

• > job before wait
• - wait
• < job after wait
• . routine does not work

async_examples.py

import concurrent
from asyncio import all_tasks, as_completed, create_task, gather, run, sleep as wait_io, wait
from collections import defaultdict
from concurrent.futures.thread import ThreadPoolExecutor
from contextlib import contextmanager
from datetime import datetime
from itertools import count
from time import sleep as blocking_job


class DashBoard:
    def __init__(self):
        self._start = datetime.now()
        self._items = defaultdict(list)
        self._id_counter = count()

    def get_next_id(self):
        return next(self._id_counter)

    def _points_from_start(self):
        return int((datetime.now() - self._start).total_seconds() * 10)

    def start(self, id_):
        self._items[id_].append(self._points_from_start())

    def end(self, id_):
        self._items[id_].append(self._points_from_start())

    def show(self):
        total_size = max(x[-1] for x in self._items.values())

        for id_, point in sorted(self._items.items()):
            line = ["."] * (total_size + 2)
            if len(point) == 2:
                start, end = point
                line[start] = ">"
                line[end] = "<"
                line[start + 1: end] = ['-'] * (end - start - 2)
            else:
                start = point[0]
                line[start] = ">"
                line[start + 1:] = ["-"] * (total_size - start)
            print(f"{id_:>2}:", "".join(line))


@contextmanager
def get_dashboard():
    dashboard = DashBoard()
    yield dashboard
    dashboard.show()


async def sleeper(io_waiting_time, dashboard: DashBoard):
    id_ = dashboard.get_next_id()
    dashboard.start(id_)
    blocking_job(0.1)
    await wait_io(io_waiting_time / 10)
    blocking_job(0.1)
    dashboard.end(id_)
    return io_waiting_time


async def run_in_order(dashboard):
    await sleeper(3, dashboard)
    await sleeper(3, dashboard)
    await sleeper(3, dashboard)


async def start_and_wait(dashboard):
    t1 = create_task(sleeper(3, dashboard))
    t2 = create_task(sleeper(3, dashboard))
    t3 = create_task(sleeper(3, dashboard))

    await t1
    await t2
    await t3


async def gather_all(dashboard):
    await gather(
        sleeper(5, dashboard),
        sleeper(1, dashboard),
        sleeper(3, dashboard)
    )


async def wait_all(dashboard):
    tasks = [

        create_task(sleeper(3, dashboard)),
        create_task(sleeper(4, dashboard)),
        create_task(sleeper(1, dashboard)),
    ]

    done, pending = await wait(tasks)


async def wait_first(dashboard):
    done, pending = await wait([
        create_task(sleeper(5, dashboard)),
        create_task(sleeper(1, dashboard)),
        create_task(sleeper(6, dashboard)),
    ], return_when=concurrent.futures.FIRST_COMPLETED)


async def make_more(dashboard):
    tasks = [
        create_task(sleeper(3, dashboard)),
        create_task(sleeper(3, dashboard)),
        create_task(sleeper(10, dashboard)),
    ]

    for _ in range(2):
        done, pending = await wait(tasks, return_when=concurrent.futures.FIRST_COMPLETED)
        for t in range(1, len(done) + 1):
            tasks = list(pending)
            tasks.append(create_task(sleeper(t, dashboard)))

    await wait(tasks, return_when=concurrent.futures.ALL_COMPLETED)


async def iterate_over_completed(dashboard):
    tasks = [
        create_task(sleeper(10, dashboard)),
        create_task(sleeper(1, dashboard)),
        create_task(sleeper(3, dashboard)),
    ]

    for future in as_completed(tasks):
        print("Returned: ", await future)


async def all_running_tasks(dashboard):
    create_task(sleeper(5, dashboard))
    create_task(sleeper(5, dashboard))
    tasks = all_tasks()
    print(len(tasks))
    await wait(tasks, timeout=1)


cases = [
    run_in_order,
    start_and_wait,
    gather_all,
    wait_all,
    wait_first,
    make_more,
    iterate_over_completed,
    all_running_tasks,
]

for function in cases:
    print(f">>> {function.__name__}\n")

    with get_dashboard() as dashboard:
        run(function(dashboard))

    print()

output.txt

>>> run_in_order

 0: >---<...........
 1: .....>---<......
 2: ..........>---<.

>>> start_and_wait

 0: >--<...
 1: .>--<..
 2: ..>--<.

>>> gather_all

 0: >----<..
 1: .>-<....
 2: ..>---<.

>>> wait_all

 0: >--<...
 1: .>---<.
 2: ..>-<..

>>> wait_first

 0: >----
 1: .>-<.
 2: ..>--

>>> make_more

 0: >--<...........
 1: .>--<..........
 2: ..>----------<.
 3: .....>--<......
 4: ......>--<.....

>>> iterate_over_completed

Returned:  1
Returned:  3
Returned:  10
 0: >---------<.
 1: .>-<........
 2: ..>---<.....

>>> all_running_tasks

3
 0: >-----<..
 1: .>-----<.