Concurrency and Parallelism

Concurrency and Parallelism#

Concurrency refers to the ability of different parts of a program to be executed out-of-order or in partial order without affecting the final outcome. Parallelism involves the simultaneous execution of multiple program parts. These concepts are crucial for improving performance in multi-core processors and distributed systems. Key topics include threading, locks, semaphores, and parallel algorithms.

👉 New to App-Generator? Sign IN with GitHub or Generate Web Apps in no time (free service).

Imagine a busy restaurant kitchen. A single chef preparing one dish at a time is like a sequential program. Multiple chefs working on different dishes simultaneously is parallelism. A single chef juggling multiple dishes by working on one while others are marinating or baking is concurrency. Let’s explore how these concepts apply to software development.

Concurrency vs. Parallelism#

  • Concurrency is about dealing with many things at once. It’s like a juggler keeping multiple balls in the air by handling them one at a time, but so quickly it appears simultaneous.

  • Parallelism is about doing many things at once. It’s like multiple jugglers each handling their own balls simultaneously.

Here’s a simple example to illustrate:

# Sequential Processing
def process_data_sequentially(items):
    for item in items:
        process_item(item)

# Concurrent Processing
async def process_data_concurrently(items):
    tasks = [process_item(item) for item in items]
    await asyncio.gather(*tasks)

# Parallel Processing
def process_data_in_parallel(items):
    with ProcessPoolExecutor() as executor:
        executor.map(process_item, items)

Threads vs. Processes#

Threads are like workers sharing the same workspace (memory). They can communicate easily but need to coordinate to avoid conflicts.

Processes are like workers in separate rooms. They have their own space (memory) and need to make explicit efforts to communicate.

import threading
import multiprocessing

# Using Threads
def thread_example():
    shared_data = []
    lock = threading.Lock()

    def worker():
        with lock:
            shared_data.append(1)

    threads = [threading.Thread(target=worker) for _ in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

# Using Processes
def process_example():
    def worker(queue):
        queue.put(1)

    queue = multiprocessing.Queue()
    processes = [multiprocessing.Process(target=worker, args=(queue,))
                for _ in range(5)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

Synchronization Mechanisms#

Locks and Mutexes#

Think of a lock as a bathroom key at a gas station - only one person can use it at a time.

import threading

class BankAccount:
    def __init__(self, balance):
        self.balance = balance
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                # Simulate some processing time
                time.sleep(0.1)
                self.balance -= amount
                return True
            return False

    def deposit(self, amount):
        with self.lock:
            # Simulate some processing time
            time.sleep(0.1)
            self.balance += amount

Semaphores#

A semaphore is like a parking lot with a limited number of spaces. It controls access to a pool of resources.

import threading

class ConnectionPool:
    def __init__(self, max_connections):
        self.semaphore = threading.Semaphore(max_connections)
        self.connections = []

    def get_connection(self):
        with self.semaphore:
            # Simulate getting a connection
            connection = create_connection()
            return connection

    def release_connection(self, connection):
        # Return connection to pool
        connection.close()
        self.semaphore.release()

Condition Variables#

Condition variables are like waiting for your order at a restaurant - you wait until the waiter notifies you that your food is ready.

import threading

class DataQueue:
    def __init__(self, size):
        self.queue = []
        self.size = size
        self.condition = threading.Condition()

    def produce(self, item):
        with self.condition:
            while len(self.queue) >= self.size:
                self.condition.wait()
            self.queue.append(item)
            self.condition.notify()

    def consume(self):
        with self.condition:
            while not self.queue:
                self.condition.wait()
            item = self.queue.pop(0)
            self.condition.notify()
            return item

Common Patterns#

Producer-Consumer Pattern#

This pattern is like a kitchen where chefs (producers) prepare food and waiters (consumers) serve it to customers.

import queue
import threading

class ProducerConsumer:
    def __init__(self):
        self.queue = queue.Queue(maxsize=10)

    def producer(self):
        while True:
            item = produce_item()
            self.queue.put(item)

    def consumer(self):
        while True:
            item = self.queue.get()
            process_item(item)
            self.queue.task_done()

Thread Pool Pattern#

A thread pool is like having a team of workers ready to handle tasks as they come in.

from concurrent.futures import ThreadPoolExecutor

class ImageProcessor:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=4)

    def process_images(self, image_files):
        futures = []
        for image in image_files:
            future = self.executor.submit(self.process_image, image)
            futures.append(future)

        # Wait for all tasks to complete
        for future in futures:
            future.result()

    def process_image(self, image):
        # Image processing logic here
        pass

Real-World Applications#

Web Server Example#

Here’s how a simple concurrent web server might handle multiple clients:

import asyncio

class WebServer:
    async def handle_client(self, reader, writer):
        request = await reader.read(100)

        # Simulate processing request
        await asyncio.sleep(0.1)

        response = self.process_request(request)
        writer.write(response)
        await writer.drain()

        writer.close()
        await writer.wait_closed()

    async def run_server(self):
        server = await asyncio.start_server(
            self.handle_client, '127.0.0.1', 8888)

        async with server:
            await server.serve_forever()

Data Processing Pipeline#

A parallel data processing pipeline using multiple processes:

from multiprocessing import Process, Queue

class DataPipeline:
    def __init__(self):
        self.input_queue = Queue()
        self.output_queue = Queue()

    def read_data(self):
        while True:
            data = read_from_source()
            self.input_queue.put(data)

    def process_data(self):
        while True:
            data = self.input_queue.get()
            result = transform_data(data)
            self.output_queue.put(result)

    def write_results(self):
        while True:
            result = self.output_queue.get()
            write_to_destination(result)

    def run(self):
        processes = [
            Process(target=self.read_data),
            Process(target=self.process_data),
            Process(target=self.write_results)
        ]

        for p in processes:
            p.start()

        for p in processes:
            p.join()

Common Challenges and Solutions#

Race Conditions#

Race conditions occur when multiple threads access shared data simultaneously. They’re like two chefs reaching for the same ingredient at the same time.

# Problematic code
class Counter:
    def __init__(self):
        self.count = 0

    def increment(self):
        # Race condition!
        self.count += 1

# Fixed version
class Counter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.count += 1

Deadlocks#

Deadlocks are like two people each waiting for the other to move out of the way in a narrow corridor.

# Potential deadlock
def transfer(account1, account2, amount):
    with account1.lock:
        with account2.lock:
            account1.withdraw(amount)
            account2.deposit(amount)

# Deadlock prevention
def transfer(account1, account2, amount):
    # Always acquire locks in a consistent order
    first = min(account1, account2, key=id)
    second = max(account1, account2, key=id)

    with first.lock:
        with second.lock:
            account1.withdraw(amount)
            account2.deposit(amount)

Best Practices#

  1. Keep It Simple: Use the simplest concurrency model that meets your needs.

  2. Avoid Shared State: Minimize shared data between threads.

  3. Use High-Level Constructs: Prefer high-level synchronization mechanisms over low-level ones.

  4. Handle Failures: Always plan for and handle potential failures in concurrent operations.

  5. Test Thoroughly: Use tools and techniques specifically designed for testing concurrent code.

Conclusion#

Concurrency and parallelism are powerful tools for improving application performance and responsiveness. The key is choosing the right approach for your specific needs and implementing it carefully to avoid common pitfalls. Remember that simpler solutions are often better - only use concurrency when it provides clear benefits and you can manage its complexity effectively.