Asyncio and Concurrency

In this chapter, we will explore the concepts of asyncio and concurrency in Python. We will cover async/await syntax, event loops, threads, processes, the Global Interpreter Lock (GIL), and the concurrent.futures module. By the end of this chapter, you will have a solid understanding of how to write concurrent code in Python and when to use each approach effectively.

Concurrency VS Parallelism

Concurrency and parallelism are related but distinct concepts in programming. Concurrency refers to the ability of a program to manage multiple tasks at the same time, while parallelism refers to the ability to execute multiple tasks simultaneously.

In Concurrency multiple tasks can be processed in overlapping time periods (context switching), while in Parallelism multiple tasks are executed at the same time (multiple CPU cores). Concurrency can be achieved through various techniques such as threading and asyncio, while parallelism can be achieved through multiprocessing.

Thread VS Process

A thread is a lightweight unit of execution that shares the same memory space with other threads in the same process. While a process is an independent unit of execution that has its own memory space. Threads are typically used for I/O-bound tasks, while processes are used for CPU-bound tasks. Threads can communicate with each other through shared memory, while processes communicate through inter-process communication (IPC) mechanisms.

Threading

Threading is a way to achieve concurrency in Python by allowing multiple threads of execution within a single process. Each thread can run concurrently, sharing the same memory space. When multiple threads execute Python code, the GIL comes into play. If those threads also access shared data, you may need a Mutex to avoid race conditions.

import threading

def task(task_name):
    for _ in range(5):
        print(f"Running {task_name}")

t1 = threading.Thread(target=task, args=("Task 1",))
t2 = threading.Thread(target=task, args=("Task 2",))

t1.start()
t2.start()
t1.join()
t2.join()

Shared Memory in Threading

Since threads share the same memory space, they can access and modify shared data. However, this can lead to race conditions if multiple threads try to modify the same data at the same time. To prevent this, you can use a Mutex (threading.Lock) to ensure that only one thread can access the shared data at a time.

import threading

counter = 0

mutex = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with mutex:  # Acquire the lock before modifying the counter
            counter += 1

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start()
t2.start()

t1.join()
t2.join()

print(f"Final counter value: {counter}")

Daemon Threads vs Non-Daemon Threads

Daemon Threads

Daemon threads are background threads that automatically terminate when the main program exits. They are typically used for tasks that should run in the background and do not need to complete before the program ends. Non-daemon threads, on the other hand, are foreground threads that keep the program running until they finish their execution.

import threading
import time

def background_task():
    while True:
        print("Running in the background...")
        time.sleep(1)

# main thread will exit after 5 seconds, and the background thread will be
# terminated automatically because it's a daemon thread.
if __name__ == "__main__":

    # Set the thread as a daemon thread
    # Note: we not join the thread because we want it to run in the
    # background and not block the main thread.
    t = threading.Thread(target=background_task, daemon=True)
    t.start()

    print("Main thread is doing some work...")
    time.sleep(5)
    print("Main thread is exiting...")

# Output:
# Main thread is doing some work...
# Running in the background...
# Running in the background...
# Running in the background...
# ...
# Main thread is exiting...

Non-Daemon Threads

Non-daemon threads will keep the program running until they finish their execution. If you create a non-daemon thread and do not join it, the program will not exit until that thread completes.

import threading
import time

def background_task():
    while True:
        print("Running in the background...")
        time.sleep(1)

if __name__ == "__main__":
    # Set the thread as a non-daemon thread (default)
    t = threading.Thread(target=background_task)
    t.start()

    print("Main thread is doing some work...")
    time.sleep(5)
    print("Main thread is exiting...")

# Output:
# Main thread is doing some work...
# Running in the background...
# Running in the background...
# Running in the background...
# ... (it will keep running indefinitely until you manually stop the program)

Multiprocessing

Multiprocessing is a way to achieve parallelism in Python by creating multiple processes. Each process has its own memory space and can run independently. This allows you to take advantage of multiple CPU cores for CPU-bound tasks.

import multiprocessing

def task(task_name):
    for _ in range(5):
        print(f"Running {task_name}")

# The __name__ == "__main__" check is required when using multiprocessing,
# Without it, each new process may re-run the entire script, causing
# infinite process creation and a RuntimeError.
if __name__ == "__main__":
    p1 = multiprocessing.Process(target=task, args=("Task 1",))
    p2 = multiprocessing.Process(target=task, args=("Task 2",))
    p1.start()
    p2.start()

    p1.join()
    p2.join()

Shared Memory in Multiprocessing

Since each process has its own memory space, they cannot directly share data. However, the multiprocessing module provides ways to share data between processes using shared memory or through inter-process communication (IPC) mechanisms like queues and pipes. When multiple processes modify the same shared memory (such as a Value or Array), locks are often needed to prevent race conditions.

We can use Queue, Value etc comes from the multiprocessing module to share data between processes. For example, we can use a Queue and Value to send data from one process to another:

# queue_example.py
from multiprocessing import Process, Queue

def worker(q):
    print("Worker is putting data in the queue")
    q.put("Hello from the worker process!")

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    print(q.get())
    p.join()

# value_example.py
from multiprocessing import Process, Value

def worker(counter):
    for _ in range(100000):
        with counter.get_lock():  # Ensure that only one process can update the counter at a time
            counter.value += 1

if __name__ == "__main__":
    counter = Value("i", 0)
    p1 = Process(target=worker, args=(counter,))
    p2 = Process(target=worker, args=(counter,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print(f"Final counter value: {counter.value}")

Deadlock

A deadlock is a situation where two or more threads or processes are waiting for each other to release resources, and as a result, none of them can proceed. This can happen when multiple threads or processes acquire locks in different orders. For example, if Thread A holds Lock 1 and waits for Lock 2, while Thread B holds Lock 2 and waits for Lock 1, both threads will be stuck waiting for each other, resulting in a deadlock.

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    with lock1:
        print("Thread 1 acquired lock 1")
        time.sleep(1)
        with lock2:
            print("Thread 1 acquired lock 2")

def thread2():
    with lock2:
        print("Thread 2 acquired lock 2")
        time.sleep(1)
        with lock1:
            print("Thread 2 acquired lock 1")

t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

t1.start()
t2.start()

t1.join()
t2.join()

# Output:
# Thread 1 acquired lock 1
# Thread 2 acquired lock 2
# (Both threads are now waiting for each other to release the locks, resulting in a deadlock)

Global Interpreter Lock (GIL)

The GIL (Global Interpreter Lock) is a special lock used by Python’s CPython interpreter. It makes sure that only one thread can execute Python code at a time. In simple terms, the GIL is a lock that allows only one thread to use the Python interpreter at a time. Other threads must wait for their turn.

Multiple threads can share the Python interpreter through context switching, but because of the GIL, only one thread can execute Python code at a time.

Imagine there is only one microphone in a room and multiple people want to speak. Even though everyone is ready to talk, only the person holding the microphone can speak. Once they are done, the microphone is passed to someone else.

The GIL works in a similar way. Even if you create multiple threads, only one thread can execute Python bytecode at a time.

import threading

def task():
    for _ in range(5):
        print("Running")

t1 = threading.Thread(target=task)
t2 = threading.Thread(target=task)

t1.start()
t2.start()

t1.join()
t2.join()

In the example above, both t1 and t2 are running, but they do not execute Python code at exactly the same time. The GIL allows one thread to run, then switches to another thread after a short period.

The GIL helps Python manage memory safely and keeps the interpreter simpler. However, it does not protect your shared variables from race conditions.

For example:

counter = 0

def increment():
    global counter
    counter += 1

If multiple threads modify counter, you can still get incorrect results. In such cases, you need a Mutex (threading.Lock) to protect the shared data.

Mutex

A Mutex (Mutual Exclusion) is simply a lock that makes sure only one thread can use a shared resource at a time.

Imagine two people trying to update the same notebook at the same time. Without a lock, both might write over each other’s changes and create a mess. A mutex acts like a key to the notebook. If one person has the key, the other must wait until the key is returned.

For example, suppose we have a variable counter that is shared between two threads (t1 and t2). Both threads want to increase the counter. If they try to do it at the same time, the final value might be incorrect. To avoid this problem, we use a mutex so that only one thread can update the counter at a time.

import threading

counter = 0
mutex = threading.Lock()

def increment():
    global counter

    for _ in range(100000):
        mutex.acquire()  # Take the lock
        counter += 1
        mutex.release()  # Give back the lock

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)

t1.start()
t2.start()

t1.join()
t2.join()

print(f"Final counter value: {counter}")

Without the mutex, both threads could update counter at the same time and get the wrong result. With the mutex, one thread updates the counter while the other waits, making the result correct and predictable.

Note

The GIL controls which thread can execute Python code, while a Mutex controls which thread can access shared data.

Asyncio

Asyncio is a Python library that helps you write asynchronous functions.

An asynchronous function is a function that can pause its work, let other tasks run, and then continue from where it stopped. This is useful when a task is waiting for something slow, such as:

• Making an API or network request
• Reading or writing a file
• Querying a database

Without asyncio, the program waits for these operations to finish before doing anything else. With asyncio, the program can work on other tasks while waiting.

Asyncio provides two main keywords:

• async - Used to define an asynchronous function.
• await - Used to pause an async function until an asynchronous operation is completed.

The await keyword can only be used inside an async function and with awaitable objects. It is commonly used for non-blocking operations.

import asyncio

async def greet():
    print("Hello")
    await asyncio.sleep(2)  # Wait for 2 seconds without blocking
    print("World")

async def main():
    await asyncio.gather(greet(), greet())  # Run two greet tasks concurrently


if __name__ == "__main__":
    asyncio.run(main())

# Output:
# Hello
# Hello
# World
# World

In this example, asyncio.sleep(2) pauses the function for 2 seconds without blocking the program. During this time, other async tasks can run.

Event Loop

The event loop is the core of asyncio. It manages and schedules the execution of asynchronous tasks. When you call asyncio.run(), it creates an event loop, runs the async function, and then closes the loop when done.

The event loop allows multiple async tasks to run concurrently. When one task is waiting (like await asyncio.sleep(2)), the event loop can switch to another task that is ready to run.

Asyncio with Threads

We can use asyncio together with threads to run blocking code without blocking the event loop. This is useful when you have tasks that involve I/O operations that can block the execution of other tasks. It is just a brief introduction to show how to use asyncio with threads. You can explore more about it in the official documentation.

concurrency.futures is a module that provides a high-level interface for asynchronously executing callables using threads or processes. It allows you to run blocking code in a separate thread or process without blocking the main event loop.

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_task():
    print("Starting blocking task...")
    time.sleep(3)  # Simulate a long-running operation
    print("Blocking task completed!")
    return "Result from blocking task"


async def main():
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        # Run the blocking task in a separate thread and await its result
        data = await loop.run_in_executor(pool, blocking_task)
    print(f"Received: {data}")

if __name__ == "__main__":
    asyncio.run(main())

# Output:
# Starting blocking task...
# Blocking task completed!
# Received: Result from blocking task

Asyncio with Processes

We can also use asyncio with processes to run CPU-bound tasks without blocking the event loop. This is useful when you have tasks that require a lot of CPU time and you want to keep the event loop responsive. It is just a brief introduction to show how to use asyncio with processes. You can explore more about it in the official documentation.

concurrency.futures is a module that provides a high-level interface for asynchronously executing callables using threads or processes. It allows you to run blocking code in a separate thread or process without blocking the main event loop.

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    print("Starting CPU-bound task...")
    total = 0
    for i in range(10**7):
        total += i
    print("CPU-bound task completed!")
    return total

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        # Run the CPU-bound task in a separate process and await its result
        result = await loop.run_in_executor(pool, cpu_bound_task)
    print(f"Result from CPU-bound task: {result}")

if __name__ == "__main__":
    asyncio.run(main())

# Output:
# Starting CPU-bound task...
# CPU-bound task completed!
# Result from CPU-bound task: 499999999500000000

Note

Here run_in_executor is used to run the blocking or CPU-bound task in a separate thread or process, allowing the event loop to remain responsive and handle other tasks concurrently.