Python Multithreading & Multiprocessing 

• Multithreading is like having multiple assistants in one kitchen (shared resources)

• Multiprocessing is like having multiple separate kitchens (independent resources)

1. Understanding Concurrency vs Parallelism

Concurrency:

• Multiple tasks appear to run simultaneously

• Tasks make progress by taking turns

• Good for I/O-bound tasks (waiting for files, network, user input)

Parallelism:

• Multiple tasks actually run simultaneously on different CPU cores

• Requires multiple processors/cores

• Good for CPU-bound tasks (calculations, data processing)

python

import time

def cpu_bound_task(n):
    """CPU-intensive task (calculations)."""
    return sum(i * i for i in range(n))

def io_bound_task(filename):
    """I/O-intensive task (waiting for I/O)."""
    time.sleep(1)  # Simulate file/network operation
    return f"Processed {filename}"

# CPU-bound = Better with multiprocessing
# I/O-bound = Better with multithreading

2. Global Interpreter Lock (GIL)

The GIL is a mutex that allows only one thread to execute Python bytecode at a time.

Implications:

• Multithreading: Only one thread executes Python code at a time

• Multiprocessing: Each process has its own GIL, so true parallelism is possible

• I/O-bound tasks: Threads can release GIL while waiting, so threading helps

• CPU-bound tasks: Threads compete for GIL, so multiprocessing is better

3. Multithreading with threading Module

Basic Threading:

python

import threading
import time

def print_numbers():
    for i in range(5):
        time.sleep(1)
        print(f"Number: {i}")

def print_letters():
    for letter in 'ABCDE':
        time.sleep(1.2)
        print(f"Letter: {letter}")

# Create threads
t1 = threading.Thread(target=print_numbers)
t2 = threading.Thread(target=print_letters)

# Start threads
t1.start()
t2.start()

# Wait for both threads to complete
t1.join()
t2.join()

print("Both threads finished!")

Threading with Arguments:

python

def worker(name, delay, count):
    """Worker function that takes arguments."""
    for i in range(count):
        time.sleep(delay)
        print(f"{name}: Task {i+1} completed")

# Create multiple threads with different arguments
threads = []
for i in range(3):
    t = threading.Thread(
        target=worker,
        args=(f"Worker-{i}", 1, 3)
    )
    threads.append(t)
    t.start()

# Wait for all threads to complete
for t in threads:
    t.join()

print("All workers finished!")

Thread Synchronization (Locks):

python

import threading

class BankAccount:
    def __init__(self, initial_balance=0):
        self.balance = initial_balance
        self.lock = threading.Lock()
    
    def deposit(self, amount):
        with self.lock:  # Acquire lock automatically
            old_balance = self.balance
            time.sleep(0.001)  # Simulate processing time
            self.balance = old_balance + amount
    
    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                old_balance = self.balance
                time.sleep(0.001)
                self.balance = old_balance - amount
                return True
            return False

def transaction_worker(account, transactions, is_deposit=True):
    for amount in transactions:
        if is_deposit:
            account.deposit(amount)
        else:
            account.withdraw(amount)

# Create account and threads
account = BankAccount(1000)

deposit_thread = threading.Thread(
    target=transaction_worker,
    args=(account, [100, 200, 300], True)
)

withdraw_thread = threading.Thread(
    target=transaction_worker, 
    args=(account, [50, 150, 250], False)
)

deposit_thread.start()
withdraw_thread.start()

deposit_thread.join()
withdraw_thread.join()

print(f"Final balance: ${account.balance}")  # Should be consistent

Thread Communication (Events and Queues):

python

x

import threading

class BankAccount:
    def __init__(self, initial_balance=0):
        self.balance = initial_balance
        self.lock = threading.Lock()
    
    def deposit(self, amount):
        with self.lock:  # Acquire lock automatically
            old_balance = self.balance
            time.sleep(0.001)  # Simulate processing time
            self.balance = old_balance + amount
    
    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                old_balance = self.balance
                time.sleep(0.001)
                self.balance = old_balance - amount
                return True
            return False

def transaction_worker(account, transactions, is_deposit=True):
    for amount in transactions:
        if is_deposit:
            account.deposit(amount)
        else:
            account.withdraw(amount)

# Create account and threads
account = BankAccount(1000)

deposit_thread = threading.Thread(
    target=transaction_worker,
    args=(account, [100, 200, 300], True)
)

withdraw_thread = threading.Thread(
    target=transaction_worker, 
    args=(account, [50, 150, 250], False)
)

deposit_thread.start()
withdraw_thread.start()

deposit_thread.join()
withdraw_thread.join()

print(f"Final balance: ${account.balance}")  # Should be consistent

4. Thread Pools with concurrent.futures

Using ThreadPoolExecutor:

python

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(task_id, duration):
    """Simulate I/O-bound task (network, file operations)."""
    print(f"Task {task_id} started")
    time.sleep(duration)  # Simulate I/O wait
    print(f"Task {task_id} completed after {duration}s")
    return f"Result from task {task_id}"

# Using ThreadPoolExecutor (better for I/O-bound tasks)
def run_with_threadpool():
    print("=== Using ThreadPoolExecutor ===")
    
    # Create thread pool with 3 workers
    with ThreadPoolExecutor(max_workers=3) as executor:
        # Submit tasks
        futures = []
        for i in range(5):
            future = executor.submit(io_task, i, 2)
            futures.append(future)
        
        # Collect results as they complete
        for future in futures:
            result = future.result()  # This blocks until result is available
            print(f"Got result: {result}")

run_with_threadpool()

Handling Exceptions in Threads:

python

x

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(task_id, duration):
    """Simulate I/O-bound task (network, file operations)."""
    print(f"Task {task_id} started")
    time.sleep(duration)  # Simulate I/O wait
    print(f"Task {task_id} completed after {duration}s")
    return f"Result from task {task_id}"

# Using ThreadPoolExecutor (better for I/O-bound tasks)
def run_with_threadpool():
    print("=== Using ThreadPoolExecutor ===")
    
    # Create thread pool with 3 workers
    with ThreadPoolExecutor(max_workers=3) as executor:
        # Submit tasks
        futures = []
        for i in range(5):
            future = executor.submit(io_task, i, 2)
            futures.append(future)
        
        # Collect results as they complete
        for future in futures:
            result = future.result()  # This blocks until result is available
            print(f"Got result: {result}")

run_with_threadpool()

5. Multiprocessing with multiprocessing Module

Basic Multiprocessing:

python

import multiprocessing
import time

def cpu_intensive_task(n, result_queue=None):
    """CPU-intensive task (better with multiprocessing)."""
    start_time = time.time()
    result = sum(i * i for i in range(n))
    end_time = time.time()
    
    if result_queue:
        result_queue.put((n, result, end_time - start_time))
    
    return result

def run_sequential():
    """Run tasks sequentially."""
    print("=== Sequential Execution ===")
    start_time = time.time()
    
    results = []
    for i in range(4):
        result = cpu_intensive_task(10**6)
        results.append(result)
    
    end_time = time.time()
    print(f"Sequential time: {end_time - start_time:.2f}s")
    return results

def run_parallel():
    """Run tasks in parallel processes."""
    print("=== Parallel Execution ===")
    start_time = time.time()
    
    processes = []
    result_queue = multiprocessing.Queue()
    
    # Create processes
    for i in range(4):
        p = multiprocessing.Process(
            target=cpu_intensive_task,
            args=(10**6, result_queue)
        )
        processes.append(p)
        p.start()
    
    # Collect results
    results = []
    for _ in range(4):
        n, result, duration = result_queue.get()
        results.append(result)
        print(f"Process completed task in {duration:.2f}s")
    
    # Wait for all processes to finish
    for p in processes:
        p.join()
    
    end_time = time.time()
    print(f"Parallel time: {end_time - start_time:.2f}s")
    return results

# Compare performance
if __name__ == "__main__":  # Important for multiprocessing on Windows
    seq_results = run_sequential()
    par_results = run_parallel()
    
    print(f"Results match: {seq_results == par_results}")

Process Pools with multiprocessing.Pool:

python

import multiprocessing

def process_task(data):
    """Task to be processed in parallel."""
    number, multiplier = data
    result = sum(i * i * multiplier for i in range(number))
    return result

def run_with_process_pool():
    """Use process pool for CPU-bound tasks."""
    print("=== Using Process Pool ===")
    
    # Data to process
    tasks = [(1000000, 1), (2000000, 2), (1500000, 1.5), (1800000, 0.8)]
    
    # Create process pool (automatically uses all available CPUs)
    with multiprocessing.Pool() as pool:
        # Map function to data (blocks until all complete)
        results = pool.map(process_task, tasks)
        
        # Alternatively, use imap for lazy evaluation
        # results = list(pool.imap(process_task, tasks))
    
    print("Results:", results)
    return results

if __name__ == "__main__":
    run_with_process_pool()

Sharing Data Between Processes:

python

import multiprocessing

def worker_with_shared_data(shared_value, shared_array, lock, worker_id):
    """Worker that modifies shared data."""
    for i in range(5):
        with lock:
            # Safely modify shared value
            shared_value.value += 1
            print(f"Worker {worker_id}: shared_value = {shared_value.value}")
        
        # Modify shared array (process-safe for simple assignments)
        shared_array[worker_id] = worker_id * 10 + i
        
        time.sleep(0.1)

def run_shared_data_example():
    """Demonstrate shared data between processes."""
    # Shared value (using Value)
    shared_value = multiprocessing.Value('i', 0)  # 'i' for integer
    
    # Shared array (using Array)
    shared_array = multiprocessing.Array('i', 4)  # Array of 4 integers
    
    # Lock for synchronization
    lock = multiprocessing.Lock()
    
    # Create processes
    processes = []
    for i in range(3):
        p = multiprocessing.Process(
            target=worker_with_shared_data,
            args=(shared_value, shared_array, lock, i)
        )
        processes.append(p)
        p.start()
    
    # Wait for all processes
    for p in processes:
        p.join()
    
    print(f"Final shared_value: {shared_value.value}")
    print(f"Final shared_array: {list(shared_array)}")

if __name__ == "__main__":
    run_shared_data_example()

6. Process Pools with concurrent.futures

Using ProcessPoolExecutor:

python

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n):
    """Check if a number is prime (CPU-intensive)."""
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False
    
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

def find_primes_in_range(start_end):
    """Find primes in a range (better for process pools)."""
    start, end = start_end
    primes = []
    for num in range(start, end + 1):
        if is_prime(num):
            primes.append(num)
    return primes

def run_prime_finder():
    """Find primes using process pool."""
    print("=== Finding Primes with ProcessPoolExecutor ===")
    
    # Ranges to check
    ranges = [(1, 10000), (10001, 20000), (20001, 30000), (30001, 40000)]
    
    start_time = time.time()
    
    with ProcessPoolExecutor() as executor:
        # Submit all tasks
        future_to_range = {
            executor.submit(find_primes_in_range, r): r 
            for r in ranges
        }
        
        # Process results as they complete
        total_primes = 0
        for future in future_to_range:
            r = future_to_range[future]
            try:
                primes = future.result()
                total_primes += len(primes)
                print(f"Range {r}: found {len(primes)} primes")
            except Exception as e:
                print(f"Range {r} failed: {e}")
    
    end_time = time.time()
    print(f"Total primes found: {total_primes}")
    print(f"Time taken: {end_time - start_time:.2f}s")

if __name__ == "__main__":
    run_prime_finder()

7. Practical Examples

Example 1: Web Scraper with Threading

python

import threading
import requests
from concurrent.futures import ThreadPoolExecutor
import time

def download_url(url):
    """Download content from a URL (I/O-bound)."""
    try:
        response = requests.get(url, timeout=10)
        return {
            'url': url,
            'status_code': response.status_code,
            'content_length': len(response.content),
            'success': True
        }
    except Exception as e:
        return {
            'url': url, 
            'error': str(e),
            'success': False
        }

def run_web_scraper(urls):
    """Scrape multiple URLs concurrently."""
    print("=== Web Scraper with ThreadPool ===")
    
    start_time = time.time()
    
    with ThreadPoolExecutor(max_workers=5) as executor:
        # Submit all download tasks
        future_to_url = {
            executor.submit(download_url, url): url 
            for url in urls
        }
        
        # Process results
        successful = 0
        failed = 0
        
        for future in future_to_url:
            url = future_to_url[future]
            try:
                result = future.result()
                if result['success']:
                    print(f"{url}: {result['status_code']} ({result['content_length']} bytes)")
                    successful += 1
                else:
                    print(f"{url}: {result['error']}")
                    failed += 1
            except Exception as e:
                print(f"{url}: Unexpected error - {e}")
                failed += 1
    
    end_time = time.time()
    print(f"\nCompleted in {end_time - start_time:.2f}s")
    print(f"Successful: {successful}, Failed: {failed}")

# Example usage
if __name__ == "__main__":
    test_urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2", 
        "https://httpbin.org/status/404",
        "https://httpbin.org/status/500",
        "https://httpbin.org/json",
        "https://invalid-website-that-does-not-exist.com",
    ]
    
    run_web_scraper(test_urls)

Example 2: Image Processor with Multiprocessing

python

import multiprocessing
import time
import math
from concurrent.futures import ProcessPoolExecutor
import os

def process_image(task_data):
    """Simulate CPU-intensive image processing."""
    image_id, operations = task_data
    
    # Simulate different processing times
    processing_time = len(operations) * 0.5
    
    # CPU-intensive work
    result = 0
    for i in range(100000 * len(operations)):
        result += math.sqrt(i) * math.sin(i)
    
    time.sleep(processing_time)  # Simulate I/O wait for file operations
    
    return {
        'image_id': image_id,
        'operations': operations,
        'processing_time': processing_time,
        'result': result,
        'process_id': os.getpid()  # Show which process handled this
    }

def run_image_processor():
    """Process multiple images in parallel."""
    print("=== Image Processor with ProcessPool ===")
    
    # Simulate image processing tasks
    tasks = [
        (1, ['resize', 'filter', 'enhance']),
        (2, ['crop', 'rotate']),
        (3, ['filter', 'compress', 'watermark']),
        (4, ['resize']),
        (5, ['enhance', 'filter', 'crop', 'compress']),
        (6, ['rotate', 'watermark']),
    ]
    
    start_time = time.time()
    
    # Use number of CPU cores
    num_processes = multiprocessing.cpu_count()
    print(f"Using {num_processes} processes")
    
    with ProcessPoolExecutor(max_workers=num_processes) as executor:
        results = list(executor.map(process_image, tasks))
    
    end_time = time.time()
    
    # Display results
    for result in results:
        print(f"Image {result['image_id']}: "
              f"{result['operations']} "
              f"(took {result['processing_time']:.1f}s, "
              f"process: {result['process_id']})")
    
    print(f"\nTotal time: {end_time - start_time:.2f}s")
    print(f"Speedup factor: {(len(tasks) * 2) / (end_time - start_time):.2f}x")

if __name__ == "__main__":
    run_image_processor()

Example 3: Real-time Data Processor

python

import threading
import queue
import time
import random
from collections import defaultdict

class RealTimeDataProcessor:
    """Process real-time data streams using producer-consumer pattern."""
    
    def __init__(self, num_workers=3):
        self.data_queue = queue.Queue()
        self.results = defaultdict(list)
        self.workers = []
        self.running = False
        self.num_workers = num_workers
        self.lock = threading.Lock()
    
    def data_producer(self, stream_id, num_messages):
        """Simulate a data stream producer."""
        for i in range(num_messages):
            # Simulate data with timestamp
            data = {
                'stream_id': stream_id,
                'timestamp': time.time(),
                'value': random.uniform(0, 100),
                'message_id': i
            }
            
            self.data_queue.put(data)
            print(f"📨 Stream {stream_id}: Produced message {i}")
            time.sleep(random.uniform(0.05, 0.2))  # Variable production rate
    
    def data_worker(self, worker_id):
        """Worker that processes data from the queue."""
        while self.running or not self.data_queue.empty():
            try:
                # Get data with timeout
                data = self.data_queue.get(timeout=1)
                
                # Simulate processing
                processing_time = random.uniform(0.1, 0.5)
                time.sleep(processing_time)
                
                # Process data (e.g., calculate moving average)
                with self.lock:
                    self.results[data['stream_id']].append(data['value'])
                
                print(f"🔧 Worker {worker_id}: Processed stream {data['stream_id']} "
                      f"message {data['message_id']} "
                      f"(took {processing_time:.2f}s)")
                
                self.data_queue.task_done()
                
            except queue.Empty:
                continue
    
    def start_processing(self, streams_config):
        """Start the data processing system."""
        print("=== Starting Real-time Data Processor ===")
        self.running = True
        
        # Start worker threads
        for i in range(self.num_workers):
            worker = threading.Thread(target=self.data_worker, args=(i,))
            self.workers.append(worker)
            worker.start()
        
        # Start producer threads for each stream
        producer_threads = []
        for stream_id, num_messages in streams_config.items():
            producer = threading.Thread(
                target=self.data_producer, 
                args=(stream_id, num_messages)
            )
            producer_threads.append(producer)
            producer.start()
        
        # Wait for all producers to finish
        for producer in producer_threads:
            producer.join()
        
        # Wait for queue to be empty
        self.data_queue.join()
        
        # Stop workers
        self.running = False
        for worker in self.workers:
            worker.join()
        
        print("=== Processing Complete ===")
        
        # Display results summary
        for stream_id, values in self.results.items():
            avg = sum(values) / len(values) if values else 0
            print(f"Stream {stream_id}: {len(values)} messages, Average: {avg:.2f}")

# Run the example
if __name__ == "__main__":
    # Configure data streams
    streams = {
        'temperature': 10,
        'pressure': 8,
        'humidity': 12
    }
    
    processor = RealTimeDataProcessor(num_workers=2)
    processor.start_processing(streams)

Key Takeaways:

When to Use What:

Best Practices:

1. Use if __name__ == "__main__" with multiprocessing (required on Windows)

2. Prefer concurrent.futures over manual thread/process management

3. Use locks when sharing data between threads

4. Use queues for thread/process communication

5. Set appropriate pool sizes (I/O: more workers, CPU: CPU count workers)

6. Handle exceptions in worker functions

7. Use context managers (with statements) for automatic cleanup