Understanding Asynchronous Execution in Single Threads

Coroutines facilitate concurrent operations within a single thread. Unlike threads, coroutines share the underlying process resources, differing only by thier private execution context stack. Switching between them occurs at the application level rather than the operating system kernel level. This minimal overhead results in higher throughput and reduced latency, particularly beneficial for applications with numerous Input/Output (I/O) operations.

Core Mechanics of Asyncio

Since Python 3.4, the asyncio library has been available as a standard module, streamlining coroutine creation. The fundamental unit is the event loop, which orchestrates the execution of scheduled tasks.

Syntax Fundamentals

To define a coroutine, use the async keyword before the function declaration. Calling such a function does not execute it immediately; instead, it returns a coroutine object.

import asyncio

coroutine_obj = None

async def execute_worker():
    print("Worker task initialized")
    await asyncio.sleep(1) # Non-blocking delay
    print("Worker task completed")
    return "Done"

The await keyword is crucial for pausing a coroutine's execution until another awaited coroutine finishes. It must be used inside an async defined functon. When await is encountered, control yields back to the event loop, allowing other pending tasks to run.

Managing the Event Loop

An event loop schedules coroutines. While modern Python often simplifies this with asyncio.run(), explicit loop management allows deeper inspection of task lifecycle.

import asyncio

async def fetch_data():
    await asyncio.sleep(0.5)
    return "Data Retrieved"

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    result = loop.run_until_complete(fetch_data())
    print(result)
finally:
    loop.close()

Alternatively, run_forever() keeps the loop active indefinitely, requiring an external signal or callback to stop execution.

Task Scheduling Strategies

The asyncio module provides several methods to manage asynchronous workflows efficiently.

• create_task(): Registers a coroutine with the loop immediately, allowing it to run concurrently.
• gather(): Executes multiple coroutines in parallel and aggregates their results into a list.
• ensure_future(): Ensures an object is wrapped in a Task.

Parallel Execution Example

Using gather allows initiating multiple independent operations simultaneously.

import asyncio

async def worker_process(identifier):
    await asyncio.sleep(2)
    print(f"Process {identifier} finished")
    return identifier

def batch_run():
    loop = asyncio.get_event_loop()
    tasks = [worker_process(i) for i in range(3)]
    
    results = loop.run_until_complete(asyncio.gather(*tasks))
    print(f"Results: {results}")
    
    loop.close()

if __name__ == '__main__':
    batch_run()

Callbacks and Future Objects

To retrieve results from tasks started via run_forever, utilize Future objects to store intermediate states and attach callbacks.

from asyncio import Future
from functools import partial

async def produce_result(task_id, future_obj):
    await asyncio.sleep(1)
    future_obj.set_result(f"Result {task_id}")

def on_completion(future, loop):
    data = future.result()
    print(data)
    loop.stop()

def start_forever_mode():
    loop = asyncio.get_event_loop()
    future = Future(loop=loop)
    
    # Create task
    asyncio.ensure_future(produce_result('A', future))
    
    # Attach callback
    future.add_done_callback(partial(on_completion, loop=loop))
    
    try:
        loop.run_forever()
    finally:
        loop.close()

if __name__ == '__main__':
    start_forever_mode()

I/O Operations and Queues

The primary advantage of async programming is handing over CPU control during I/O waits. This applies to both network requests and file operations.

Network Interaction

For asynchronous networking, libraries like aiohttp are preferred over synchronous ones like requests. They integrate directly with the event loop.

import asyncio
import aiohttp

async def request_resource(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            content = await response.read()
    return len(content)

async def main_network():
    urls = ['https://example.com'] * 10
    start_time = asyncio.get_event_loop().time()
    
    tasks = [request_resource(u) for u in urls]
    await asyncio.gather(*tasks)
    
    end_time = asyncio.get_event_loop().time()
    print(f"Duration: {end_time - start_time:.2f}s")

Producer-Consumer Model

Synchronization between coroutines is often handled via Queues. asyncio.Queue supports FIFO patterns but lacks thread safety for multi-process environments.

import asyncio

async def producer(q: asyncio.Queue):
    for i in range(5):
        await q.put(i)
        print(f"Produced: {i}")

async def consumer(q: asyncio.Queue):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        q.task_done()

def orchestrate_queue():
    loop = asyncio.get_event_loop()
    q = asyncio.Queue()
    
    # Signal finish logic could go here
    
    loop.run_until_complete(asyncio.gather(
        producer(q),
        consumer(q)
    ))
    loop.close()

if __name__ == '__main__':
    orchestrate_queue()

Key Classes

• Future: Represents a placeholder for a value that may not yet be available.
• Task: A Future subclass wrapping a coroutine to execute it on the loop.

Standard attributes include result() for fetching values and done() to check status.