Python Multithreading

Python Concurrency: Threading

Concurrency allows multiple tasks to run or manage execution in overlapping time periods. Python offers several ways to achieve concurrency, with threading being one common approach.

Benefits of using threads:

Shared Memory: Threads within the same process share the same memory space, making communication and data sharing between them relatively straightforward compared to separate processes (multiprocessing).
Resource Efficiency: Threads (often called lightweight processes) generally consume fewer system resources (memory, initialization time) than full processes.
Responsiveness: Useful for keeping applications responsive (e.g., handling GUI events while performing background tasks) or managing multiple I/O-bound operations concurrently (like network requests or file operations).

What is a Thread?

A thread is the smallest sequence of programmed instructions that can be managed independently by a scheduler. A process can contain multiple threads, all executing within the same process context.

Each thread has its own execution context (including an instruction pointer and registers).
Threads can be paused (put to sleep) or yield control, allowing other threads to run.
The operating system scheduler manages switching between threads (pre-emptive multitasking).

The Global Interpreter Lock (GIL)

Crucially, in the standard CPython interpreter, the Global Interpreter Lock (GIL) ensures that only one thread executes Python bytecode at a time within a single process, even on multi-core processors. This simplifies memory management but limits true parallelism for CPU-bound tasks.

Implications of the GIL:

Threading is effective for I/O-bound tasks (where threads spend time waiting for external operations like network or disk I/O). While one thread waits, the GIL can be released, allowing another thread to run.
Threading does not provide speedup for CPU-bound tasks (tasks performing intensive calculations) in CPython, as only one thread runs Python code at any moment.
For CPU-bound parallelism, use the multiprocessing module, which creates separate processes, each with its own interpreter and memory space, bypassing the GIL.
The asyncio library provides an alternative concurrency model using an event loop, suitable for high-level structured network code and other I/O-bound tasks, often more efficient than threading for managing very large numbers of concurrent connections.

The `threading` Module (Recommended)

The threading module is the modern, high-level interface for working with threads in Python. It builds upon the lower-level (and now legacy) _thread module (formerly thread in Python 2). Always prefer threading over _thread for new code.

Key Components of `threading`:

threading.Thread: The class representing a thread of execution.
threading.Lock, threading.RLock: Synchronization primitives to prevent race conditions when multiple threads access shared resources.
threading.Event: A simple mechanism for communication between threads (one thread signals an event, others wait for it).
threading.Condition, threading.Semaphore, threading.Barrier: More advanced synchronization tools.
threading.current_thread(): Returns the current Thread object.
threading.active_count(): Returns the number of active Thread objects.

Creating and Starting Threads

There are two main ways to create threads using threading.Thread:

1. Passing a Callable (Function) to the Constructor:

This is often the simpler approach for straightforward tasks.

#!/usr/bin/env python3

import threading
import time

def print_time(thread_name, delay):
    """Prints the current time every 'delay' seconds, 5 times."""
    count = 0
    while count < 5:
        time.sleep(delay)
        count += 1
        current_time = time.strftime("%H:%M:%S", time.localtime())
        print(f"{thread_name}: {current_time} (Count: {count})")
    print(f"{thread_name} exiting.")

# Create threads by passing the target function and arguments
try:
    thread1 = threading.Thread(target=print_time, args=("Thread-1", 1))
    thread2 = threading.Thread(target=print_time, args=("Thread-2", 2))

    # Start the new threads
    thread1.start()
    thread2.start()

    # Wait for both threads to complete before the main thread exits
    thread1.join()
    thread2.join()

except Exception as e:
    print(f"Error: unable to start thread - {e}")

print("Main thread exiting.")

2. Subclassing `threading.Thread`:

Override the __init__ method (optional, call parent __init__) and the run() method (required, contains the thread’s execution logic).

#!/usr/bin/env python3

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, thread_id, name, delay):
        # Call the Thread class's initializer
        super().__init__()
        # Or: threading.Thread.__init__(self)
        self.thread_id = thread_id
        self.name = name
        self.delay = delay

    def run(self):
        """This method is executed when thread.start() is called."""
        print(f"Starting {self.name}")
        # Call the function containing the thread's main logic
        print_task(self.name, self.delay)
        print(f"Exiting {self.name}")

def print_task(thread_name, delay):
    """A simple task for the thread to perform."""
    count = 0
    while count < 3:
        time.sleep(delay)
        count += 1
        current_time = time.strftime("%H:%M:%S", time.localtime())
        print(f"{thread_name}: {current_time} (Count: {count})")

# Create new threads by instantiating the subclass
thread1 = MyThread(1, "Thread-A", 1)
thread2 = MyThread(2, "Thread-B", 1.5)

# Start new Threads (this calls the run() method)
thread1.start()
thread2.start()

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

print("Main thread exiting.")

Key `Thread` Methods:

start(): Starts the thread’s activity. It calls the run() method (or the target function) in a separate thread of control.
run(): The method representing the thread’s activity. Override this in subclasses.
join([timeout]): Waits until the thread terminates. Blocks the calling thread. If timeout is given, it waits at most that many seconds.
is_alive(): Returns True if the thread is still executing, False otherwise.
name / getName() / setName(): Get or set the thread’s name.
ident: The thread’s unique identifier (or None if not started).

Synchronizing Threads (`Lock`)

When multiple threads access shared resources (like variables or data structures), race conditions can occur, leading to unpredictable results. Synchronization primitives like locks are used to control access.

A threading.Lock provides mutual exclusion. Only one thread can ‘acquire’ the lock at a time. Other threads attempting to acquire it will block until the lock is ‘released’.

Using Locks with `with` Statement (Recommended):

The with statement simplifies lock usage by automatically acquiring the lock upon entering the block and releasing it upon exiting, even if errors occur.

#!/usr/bin/env python3

import threading
import time

shared_counter = 0
counter_lock = threading.Lock() # Create a lock object

class WorkerThread(threading.Thread):
    def __init__(self, name):
        super().__init__(name=name)

    def run(self):
        global shared_counter
        print(f"{self.name} starting.")
        for _ in range(100000):
            # Acquire the lock before accessing the shared resource
            with counter_lock:
                # --- Critical Section Start ---
                current_value = shared_counter
                # Simulate some processing time
                time.sleep(0.000001)
                shared_counter = current_value + 1
                # --- Critical Section End ---
            # Lock is automatically released here
        print(f"{self.name} finished.")

# Create threads
thread1 = WorkerThread("Worker-1")
thread2 = WorkerThread("Worker-2")

# Start threads
thread1.start()
thread2.start()

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

print(f"Main thread exiting. Final counter value: {shared_counter}")
# Without the lock, the final value would likely be less than 200000 due to race conditions.

Manual Lock Usage (Less Recommended):

You can manually call lock.acquire() and lock.release(), but you must ensure release() is called, typically using a try...finally block.

# Inside the run method (alternative to 'with')
counter_lock.acquire()
try:
    # Access shared resource
    current_value = shared_counter
    shared_counter = current_value + 1
finally:
    counter_lock.release() # Ensure lock is always released

Using Queues for Thread Communication (`queue` module)

The queue module (note lowercase ‘q’ in Python 3) provides thread-safe queue classes, ideal for passing data between producer and consumer threads.

Common Queue Methods:

queue.Queue(maxsize=0): Creates a FIFO (First-In, First-Out) queue. maxsize=0 means infinite size.
put(item, block=True, timeout=None): Adds item to the queue. If block is True and the queue is full (if maxsize > 0), waits until space is available (or timeout expires).
get(block=True, timeout=None): Removes and returns an item from the queue. If block is True and the queue is empty, waits until an item is available (or timeout expires).
qsize(): Returns the approximate number of items in the queue.
empty(): Returns True if the queue is empty, False otherwise.
full(): Returns True if the queue is full, False otherwise.
task_done(): Call this after processing an item retrieved with get(). Used with join().
join(): Blocks until all items in the queue have been retrieved and processed (i.e., task_done() called for each item).

Other queue types: LifoQueue (Last-In, First-Out), PriorityQueue.

Example: Producer-Consumer Pattern

#!/usr/bin/env python3

import queue
import threading
import time

# A thread-safe queue
work_queue = queue.Queue(10)
exit_flag = False

class ProducerThread(threading.Thread):
    def __init__(self, name, q):
        super().__init__(name=name)
        self.q = q

    def run(self):
        print(f"{self.name} starting.")
        for i in range(1, 6):
            if exit_flag:
                break
            item = f"Data_{i}"
            self.q.put(item) # Add item to the queue
            print(f"{self.name} produced {item}")
            time.sleep(0.5)
        print(f"{self.name} finished producing.")

class ConsumerThread(threading.Thread):
    def __init__(self, name, q):
        super().__init__(name=name)
        self.q = q

    def run(self):
        print(f"{self.name} starting.")
        while not exit_flag or not self.q.empty():
            try:
                # Get item from queue, wait up to 1 sec if empty
                item = self.q.get(block=True, timeout=1)
                print(f"{self.name} consuming {item}")
                # Simulate processing time
                time.sleep(1)
                self.q.task_done() # Signal that processing is complete
            except queue.Empty:
                if exit_flag:
                    break # Exit if flag is set and queue is empty
                else:
                    continue # Queue is temporarily empty, try again
            except Exception as e:
                 print(f"{self.name} encountered error: {e}")
                 self.q.task_done() # Still mark task done if error occurs
        print(f"{self.name} finished consuming.")


# Create threads
producer = ProducerThread("Producer", work_queue)
consumer1 = ConsumerThread("Consumer-1", work_queue)
consumer2 = ConsumerThread("Consumer-2", work_queue)

# Start threads
producer.start()
consumer1.start()
consumer2.start()

# Wait for producer to finish
producer.join()

# Wait for the queue to be empty (all items processed)
print("Waiting for consumers to finish...")
work_queue.join() # Blocks until task_done() called for all items

# Signal consumers to exit
print("Setting exit flag.")
exit_flag = True

# Wait for consumers to exit cleanly
consumer1.join()
consumer2.join()

print("Main thread exiting.")

Python Multithreading

Python Concurrency: Threading

What is a Thread?

The Global Interpreter Lock (GIL)

The threading Module (Recommended)

Key Components of threading: