<jupyter_start><jupyter_text>Helper functions<jupyter_code>def print_with_elapsed_time(msg: str):
    global current_time
    """Print message and elapsed seconds since current_time was reset."""
    print(f'{time.perf_counter() - current_time:.1f}s: {msg}')


def run(func):
    """Run async function in existing (Jupyter) or a new loop."""
    global current_time
    current_time = time.perf_counter()
    print(f'\n--------- running {func.__name__} ---------')
    loop = asyncio.get_event_loop()
    if loop.is_running():
        loop.create_task(func())
    else:
        loop.run_until_complete(func())<jupyter_output><empty_output><jupyter_text>Single async task<jupyter_code>async def async_work_1():
    print_with_elapsed_time("Starting A")
    await asyncio.sleep(1)
    print_with_elapsed_time("Finished A")
run(async_work_1)<jupyter_output>--------- running async_work_1 ---------
0.0s: Starting A
1.0s: Finished A<jupyter_text>Single async task with result<jupyter_code>async def async_work_2(n: float):
    print_with_elapsed_time(f"Starting B with n={n}")
    await asyncio.sleep(1)
    print_with_elapsed_time(f"Finished B with n={n}")
    return n


async def task_with_result():
    result = await async_work_2(n=100)
    print_with_elapsed_time(f'result={result}')
run(task_with_result)<jupyter_output>--------- running task_with_result ---------
0.0s: Starting B with n=100
1.0s: Finished B with n=100
1.0s: result=100<jupyter_text>Multiple async tasks with resultsUse `asyncio.gather()` to run tasks in parallel.<jupyter_code>async def multiple_tasks_with_results():
    result = await asyncio.gather(
        async_work_1(),
        async_work_2(n=100),
        async_work_2(n=200)
    )
    print_with_elapsed_time(f'result={result}')
run(multiple_tasks_with_results)<jupyter_output>--------- running multiple_tasks_with_results ---------
0.0s: Starting A
0.0s: Starting B with n=100
0.0s: Starting B with n=200
1.0s: Finished A
1.0s: Finished B with n=100
1.0s: Finished B with n=200
1.0s: result=[None, 100, 200]<jupyter_text>Task which includes blocking call`work_1()` will delay the whole processing for 1.5s,since it uses blocking `time.sleep()` call.  The timing will depend on the order of `asyncio.gather()` parameters.<jupyter_code>def work_1(n: int):
    print_with_elapsed_time(f"Starting blocking work with n={n}")
    time.sleep(1.5)
    print_with_elapsed_time(f"Finished blocking work with n={n}")
    return n


async def async_work_3(n: int):
    print_with_elapsed_time("Starting C")
    work_1(n)
    print_with_elapsed_time("Finished C")


async def blocking_task():
    result = await asyncio.gather(
        async_work_3(n=100),
        async_work_1(),
        async_work_2(n=200)
    )
    print_with_elapsed_time(f'result={result}')
run(blocking_task)<jupyter_output>--------- running blocking_task ---------
0.0s: Starting C
0.0s: Starting blocking work with n=100
1.5s: Finished blocking work with n=100
1.5s: Finished C
1.5s: Starting A
1.5s: Starting B with n=200
2.5s: Finished A
2.5s: Finished B with n=200
2.5s: result=[None, None, 200]<jupyter_text>Convert blocking function into non-blockingUse `run_in_executor()` to convert blocking into non-blocking tasks:- I/O-bound tasks should be run on `ThreadPoolExecutor`- CPU-bound tasks should be run on `ProcessPoolExecutor` (to avoid GIL limitations)<jupyter_code>async def async_work_4(n: int, executor):
    print_with_elapsed_time("Starting D")
    loop = asyncio.get_running_loop()
    if executor is None:
        result = await loop.run_in_executor(None, work_1, n)  # default executor
    else:
        with executor() as pool:
            result = await loop.run_in_executor(pool, work_1, n)
    print_with_elapsed_time("Finished D")
    return result


async def blocking_task_on_default_executor():
    result = await asyncio.gather(
        async_work_4(100, executor=None),
        async_work_1(),
        async_work_2(n=200)
    )
    print_with_elapsed_time(f'result={result}')


async def blocking_task_on_thread_executor():
    result = await asyncio.gather(
        async_work_4(100, executor=ThreadPoolExecutor),
        async_work_1(),
        async_work_2(n=200)
    )
    print_with_elapsed_time(f'result={result}')


async def blocking_task_on_process_executor():
    result = await asyncio.gather(
        async_work_4(100, executor=ProcessPoolExecutor),
        async_work_1(),
        async_work_2(n=200)
    )
    print_with_elapsed_time(f'result={result}')
run(blocking_task_on_default_executor)
run(blocking_task_on_thread_executor)
run(blocking_task_on_process_executor)<jupyter_output>--------- running blocking_task_on_process_executor ---------
0.0s: Starting D
0.0s: Starting blocking work with n=100
0.0s: Starting A
0.0s: Starting B with n=200
1.0s: Finished A
1.0s: Finished B with n=200
1.5s: Finished blocking work with n=100
1.6s: Finished D
1.6s: result=[100, None, 200]
{'fraction_non_alphanumeric': 0.12329603255340793, 'fraction_numerical': 0.038046795523906406, 'mean_word_length': 4.104880581516095, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '215036', 'license': 'MIT', 'n_tokens_mistral': 1809, 'n_tokens_neox': 1723, 'n_words': 395, 'path': 'notebooks/async_tasks.ipynb', 'repo_name': 'blazzup/python-notes', 'size': '4915'}