xbe.at

Python / Performance and concurrency

Concurrency and Parallelism in Python Slideshow Guide

7 min read

Introduction to Concurrency and Parallelism

Concurrency and parallelism are fundamental concepts in modern computing, enabling efficient utilization of system resources and improved performance. While concurrency deals with managing multiple tasks or threads of execution, parallelism involves executing multiple tasks simultaneously on different processors or cores.

import threading
import time

def worker():
    print(f"Worker thread started: {threading.current_thread().name}")
    time.sleep(2)
    print(f"Worker thread finished: {threading.current_thread().name}")

# Create and start two worker threads
thread1 = threading.Thread(target=worker)
thread2 = threading.Thread(target=worker)
thread1.start()
thread2.start()

print("Main thread waiting for worker threads to finish...")

# Wait for the worker threads to complete
thread1.join()
thread2.join()

print("All worker threads have finished.")

Threading in Python

Python's threading module enables the creation and management of threads within a single process. Threads allow concurrent execution of multiple tasks, sharing the same memory space. However, Python's Global Interpreter Lock (GIL) limits true parallelism for CPU-bound tasks.

import threading

# Define a function to be executed by the thread
def print_numbers():
    for i in range(10):
        print(i)

# Create and start two threads
thread1 = threading.Thread(target=print_numbers)
thread2 = threading.Thread(target=print_numbers)

thread1.start()
thread2.start()

# Wait for the threads to complete
thread1.join()
thread2.join()

print("Done!")

Multiprocessing in Python

Python's multiprocessing module allows the creation of separate processes, each with its own memory space and GIL. This enables true parallelism for CPU-bound tasks, as processes can run concurrently on multiple cores or processors.

from multiprocessing import Pool

def square(x):
    return x ** 2

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        result = pool.map(square, [1, 2, 3, 4, 5])
        print(result)

I/O-bound vs. CPU-bound Tasks

I/O-bound tasks involve waiting for input/output operations, such as reading from a file or network. CPU-bound tasks involve intensive computational work. Python's threading is suitable for I/O-bound tasks, while multiprocessing is better for CPU-bound tasks.

import time
import threading

# I/O-bound task
def read_file(file_path):
    with open(file_path, 'r') as file:
        data = file.read()
    print(f"File '{file_path}' read successfully.")

# CPU-bound task
def compute_sum(n):
    total = sum(range(n))
    print(f"Sum of numbers up to {n} is {total}.")

# Create threads for I/O-bound tasks
thread1 = threading.Thread(target=read_file, args=('file1.txt',))
thread2 = threading.Thread(target=read_file, args=('file2.txt',))

# Create processes for CPU-bound tasks
process1 = multiprocessing.Process(target=compute_sum, args=(10000000,))
process2 = multiprocessing.Process(target=compute_sum, args=(20000000,))

# Start threads and processes
thread1.start()
thread2.start()
process1.start()
process2.start()

# Wait for threads and processes to complete
thread1.join()
thread2.join()
process1.join()
process2.join()

Thread Synchronization

When multiple threads access shared resources concurrently, race conditions and data corruption can occur. Python provides various synchronization primitives, such as locks, semaphores, and events, to ensure thread safety.

import threading

# Shared resource
counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

# Create and start two threads
thread1 = threading.Thread(target=increment)
thread2 = threading.Thread(target=increment)

thread1.start()
thread2.start()

# Wait for threads to complete
thread1.join()
thread2.join()

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

Thread Pools

Thread pools provide a way to manage and reuse a pool of worker threads, reducing the overhead of creating and destroying threads for each task. Python's concurrent.futures module offers a high-level interface for working with thread pools.

import concurrent.futures

def square(x):
    return x ** 2

if __name__ == "__main__":
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        numbers = [1, 2, 3, 4, 5]
        results = [executor.submit(square, n) for n in numbers]

        for future in concurrent.futures.as_completed(results):
            result = future.result()
            print(result)

Event-Driven Programming with Asyncio

Python's asyncio module provides an event loop and coroutines for writing concurrent code using the async/await syntax. It allows efficient handling of I/O-bound tasks and enables cooperative multitasking.

import asyncio

async def fetch_data(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(2)  # Simulating I/O operation
    return f"Data from {url}"

async def main():
    urls = ["http://example.com", "http://python.org", "http://google.com"]
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

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

Producer-Consumer Pattern

The producer-consumer pattern is a common concurrency design pattern where one or more producer threads generate data, and one or more consumer threads process the data. Python's queue module provides a thread-safe implementation of queues for inter-thread communication.

import threading
import queue
import random
import time

# Producer thread
def producer(q, event):
    while not event.is_set():
        data = random.randint(1, 100)
        q.put(data)
        print(f"Produced: {data}")
        time.sleep(1)

# Consumer thread
def consumer(q, event):
    while not event.is_set() or not q.empty():
        try:
            data = q.get(timeout=1)
            print(f"Consumed: {data}")
        except queue.Empty:
            continue

# Create a queue and an event
q = queue.Queue()
event = threading.Event()

# Create and start producer and consumer threads
producer_thread = threading.Thread(target=producer, args=(q, event))
consumer_thread = threading.Thread(target=consumer, args=(q, event))

producer_thread.start()
consumer_thread.start()

# Wait for some time before stopping the threads
time.sleep(10)
event.set()

# Wait for threads to finish
producer_thread.join()
consumer_thread.join()

Parallel Processing with Multiprocessing Pool

The multiprocessing module in Python provides a Pool class for parallel execution of tasks across multiple processes. It simplifies the creation and management of a pool of worker processes, automatically distributing tasks among them.

import multiprocessing

def square(x):
    return x ** 2

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=4)
    numbers = [1, 2, 3, 4, 5, 6, 7, 8]
    results = pool.map(square, numbers)
    print(results)
    pool.close()
    pool.join()

Inter-Process Communication

When working with multiple processes, communication between them is often necessary. Python's multiprocessing module provides various synchronization primitives, such as queues, pipes, and shared memory, to facilitate inter-process communication and data sharing.

from multiprocessing import Process, Queue

def producer(q):
    for i in range(10):
        q.put(i)
        print(f"Producer put {i} in the queue.")

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumer got {item} from the queue.")

if __name__ == "__main__":
    queue = Queue()
    producer_process = Process(target=producer, args=(queue,))
    consumer_process = Process(target=consumer, args=(queue,))

    producer_process.start()
    consumer_process.start()

    producer_process.join()
    queue.put(None)  # Signal the consumer to stop
    consumer_process.join()

Distributed Computing with Dask

Dask is a parallel computing library for Python that scales from multi-core machines to distributed clusters. It provides high-level APIs for parallel data processing, allowing you to write code once and run it in various environments, including multi-core, distributed, or GPU-accelerated systems.

import dask.bag as db

# Create a Dask Bag from a list of numbers
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
bag = db.from_sequence(numbers)

# Perform parallel operations
squared = bag.map(lambda x: x ** 2)
total = squared.sum().compute()

print(f"Sum of squared numbers: {total}")

Concurrent File Operations

Python's concurrent.futures module provides a high-level interface for working with thread and process pools, making it easier to execute concurrent file operations, such as reading, writing, or processing multiple files simultaneously.

import concurrent.futures
import os

def process_file(file_path):
    with open(file_path, 'r') as file:
        data = file.read()
        # Perform some processing on the data
        processed_data = data.upper()
        return processed_data

if __name__ == "__main__":
    file_paths = ['file1.txt', 'file2.txt', 'file3.txt']

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = [executor.submit(process_file, path) for path in file_paths]

        for future in concurrent.futures.as_completed(futures):
            processed_data = future.result()
            print(processed_data)

Asynchronous Web Scraping

Python's aiohttp and asyncio libraries enable asynchronous web scraping, allowing concurrent fetching of web pages without blocking the main thread. This approach can significantly improve performance when scraping multiple URLs.

import aiohttp
import asyncio

async def fetch_url(url, session):
    async with session.get(url) as response:
        html = await response.text()
        print(f"Fetched {url}: {len(html)} characters")
        return html

async def main():
    urls = [
        "http://example.com",
        "http://python.org",
        "http://google.com"
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch_url(url, session)) for url in urls]
        await asyncio.gather(*tasks)

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

Additional Resources

For further learning and exploring concurrency and parallelism in Python, here are some recommended resources from arXiv.org:

  • "Parallel and Concurrent Programming in Python" by Giancarlo Zizzi (arXiv:1609.03609)
  • "Concurrent Programming with Python: From Theory to Practice" by Shahram Rahatlou (arXiv:1805.04092)
  • "Parallel and Distributed Computing with Python" by Jeff Albrecht (arXiv:1907.10556)

arXiv links:

More in Performance and concurrency

  1. Disabling GIL in Python 3.13 for Faster Multi-Threading4 min

    Python 3.13 introduces a significant change in how it handles multi-threading. The Global Interpreter Lock (GIL), a mechanism that prevents multiple native...

  2. Accelerating Python with Numba9 min

    Numba is a just-in-time compiler for Python that can significantly speed up numerical and scientific Python code. It works by translating Python functions to...

  3. Disabling the Python Global Interpreter Lock (GIL)5 min

    The Global Interpreter Lock (GIL) is a mechanism used in CPython, the reference implementation of Python, to synchronize thread execution. It ensures that...

  4. 3 Approaches to Concurrency in Python6 min

    Concurrency in Python allows multiple tasks to run seemingly simultaneously, improving performance and efficiency. This slideshow explores three main...

  5. Efficient Asynchronous Socket Programming with asyncio in Python11 min

    Understanding asynchronous socket programming requires grasping how asyncio manages non-blocking I/O operations. The asyncio library provides high-level APIs...

  6. Mastering Python Memory Management! Profiling and Optimization4 min

    Python's memory management is automatic, but understanding its intricacies can help optimize your code. This presentation will cover profiling techniques and...