Threading the Needle: A Quirky Guide to Python's Concurrent Programming Tools

I was trying to see what were the threading primitives that python had to offer and thought it would be an interesting blog post to use them all.

Let's write a quirky story for each.

Thread: The Parallel Pizza Party

Imagine you're hosting a pizza party, but you only have one phone to call for delivery. In the old days, you'd call one pizza place, wait for the delivery, then call the next place. Your guests would be waiting forever!

With threads, it's like having multiple phones. You can call "Pizza Palace" for pepperoni and "Cheesy Delights" for cheese pizza simultaneously. Both orders are processed at the same time, and your pizzas arrive much faster.

In our code, each thread is like a separate phone call, downloading a different file simultaneously. The .join() method is like waiting for all pizza deliveries to arrive before starting the party.

import threading
import time

def download_file(file_name):
    print(f"Starting download of {file_name}...")
    # Simulate file download
    time.sleep(2)
    print(f"Download of {file_name} complete!")

# Create threads for downloading multiple files simultaneously
thread1 = threading.Thread(target=download_file, args=("data1.csv",))
thread2 = threading.Thread(target=download_file, args=("data2.csv",))

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

# Wait for both downloads to complete
thread1.join()
thread2.join()

print("All downloads completed!")

Lock: The Single Bathroom Dilemma

Picture a house party with only one bathroom. Without any system, chaos ensues as people try to use it simultaneously (awkward!). The solution? A simple lock on the door.

When someone needs the bathroom, they check if it's available. If it is, they lock the door, do their business, and then unlock it when they leave. If it's occupied, they wait until it's free.

import threading
import time

# Shared resource - bank account
account_balance = 1000
lock = threading.Lock()

def make_withdrawal(amount):
    global account_balance
    
    # Acquire lock before accessing shared resource
    lock.acquire()
    try:
        if account_balance >= amount:
            # Simulate processing time
            time.sleep(0.1)
            account_balance -= amount
            print(f"Withdrew ${amount}. Remaining balance: ${account_balance}")
        else:
            print(f"Failed to withdraw ${amount}. Insufficient funds.")
    finally:
        # Always release the lock
        lock.release()

RLock: The Nested Meeting Room

Imagine you're a manager who books a conference room for a team meeting. During that meeting, you realize you need a private conversation with one team member, so you book the same room for a one-on-one right after. With a regular lock, you'd have to end the team meeting, release the room, and then re-book it. That's inefficient!

An RLock (Reentrant Lock) is like having special booking privileges - you can "book" the same room multiple times without releasing it first, as long as you're the one who booked it originally.

import threading

class FileManager:
    def __init__(self):
        self.lock = threading.RLock()  # Reentrant lock
        self.file_data = {}
    
    def update_file(self, file_name, content):
        with self.lock:
            print(f"Updating {file_name}")
            # This method calls another method that also acquires the lock
            self._write_to_file(file_name, content)
    
    def _write_to_file(self, file_name, content):
        # With RLock, this can acquire the lock again without deadlock
        with self.lock:
            self.file_data[file_name] = content
            print(f"Written to {file_name}: {content}")

In our code, the update_file method acquires the lock, then calls _write_to_file, which also tries to acquire the lock. With an RLock, this works fine because the same thread can acquire the lock multiple times.

Condition: The Coffee Shop Conundrum

Picture a busy coffee shop with baristas (producers) making coffee and customers (consumers) waiting for their orders. When there are no coffees ready, customers wait. When a coffee is ready, the barista calls out "Order up!" and a waiting customer grabs it. If all the pickup counter is full, baristas wait until there's space before making more coffees.

A Condition variable is like this coffee shop system - it allows threads to wait until a specific condition is met, and then be notified when it is.

import threading
import time
import random

# Simulate a producer-consumer pattern for a message queue
message_queue = []
MAX_QUEUE_SIZE = 5
condition = threading.Condition()

def producer():
    global message_queue
    for i in range(10):
        # Acquire the condition lock
        with condition:
            # Wait if the queue is full
            while len(message_queue) >= MAX_QUEUE_SIZE:
                print("Queue full, producer waiting...")
                condition.wait()
            
            # Add a message to the queue
            message = f"Message-{i}"
            message_queue.append(message)
            print(f"Produced: {message}")
            
            # Notify consumers that a new message is available
            condition.notify()

In our code, the producer waits when the queue is full and notifies consumers when a new message is available. Consumers wait when the queue is empty and notify producers when they've consumed a message, creating a coordinated dance of production and consumption.

Semaphore: The Limited Pool Passes

Imagine a community pool with only three swimming lanes. The lifeguard gives out exactly three passes. When you want to swim, you must get a pass. If all passes are taken, you wait until someone finishes swimming and returns their pass.

A Semaphore is like this pass system - it allows a specific number of threads to access a resource simultaneously.

import threading
import time
import random

# Simulate a connection pool with limited connections
class DatabaseConnectionPool:
    def __init__(self, max_connections=3):
        # BoundedSemaphore ensures we never release more than we acquire
        self.connection_semaphore = threading.BoundedSemaphore(max_connections)
        self.connections = [f"Connection-{i}" for i in range(max_connections)]
        self.lock = threading.Lock()

In our code, the BoundedSemaphore ensures that only three database connections can be used at once. If all connections are in use, new requests wait until a connection becomes available. The "bounded" part ensures we never accidentally create more than three connections, just like the lifeguard would never hand out a fourth swimming pass.

Event: The Grand Opening

Picture a new store's grand opening. Customers line up outside, waiting for the doors to open. The store manager (the main thread) is inside preparing everything. When everything is ready, the manager flips the "OPEN" sign (sets the event), and all the waiting customers can enter at once.

An Event is like this "OPEN" sign - it allows multiple threads to wait until a specific event occurs, then all proceed once it does.

import threading
import time
import random

# Simulate a system startup with dependent services
system_ready = threading.Event()

def service_startup(service_name, startup_time):
    print(f"{service_name} is starting up...")
    time.sleep(startup_time)  # Simulate startup time
    print(f"{service_name} has started successfully!")
    
    if service_name == "Database":
        # Database is the critical service, signal when it's ready
        print("Critical service is online, system can now process requests")
        system_ready.set()  # Set the event flag to True

In our code, client threads wait for the system to be ready (the "OPEN" sign). Once the database service is up, it sets the event, allowing all waiting clients to proceed with their requests simultaneously.

Timer: The Absent-Minded Professor

Meet Professor Forgetful, who always gets so absorbed in his research that he forgets to save his work. His clever assistant set up an automatic reminder that pops up every 5 minutes saying, "Professor, save your work!"

A Timer is like this automatic reminder - it executes a function after a specified delay, without blocking the main program.

import threading
import time

def auto_save(document_name):
    print(f"Auto-saving document: {document_name}")
    # Schedule the next auto-save in 5 seconds
    timer = threading.Timer(5.0, auto_save, args=(document_name,))
    timer.daemon = True  # Allow the program to exit even if timer is alive
    timer.start()

In our code, the auto-save function runs, then schedules itself to run again in 5 seconds, creating a recurring reminder that saves the document while the user continues working.

Barrier: The Synchronized Swimmers

Imagine a team of synchronized swimmers. Each swimmer performs their individual routine, but at certain points, they all need to meet in the center of the pool before starting the next sequence together.

A Barrier is like this synchronization point - it ensures that all threads reach a certain point before any of them proceed to the next step.

import threading
import time
import random

def simulate_distributed_calculation(worker_id, barrier, results):
    print(f"Worker {worker_id} starting phase 1 calculation...")
    # Simulate phase 1 calculation
    time.sleep(random.uniform(1, 3))
    results[worker_id] = random.randint(1, 100)
    print(f"Worker {worker_id} finished phase 1 with result: {results[worker_id]}")
    
    # Wait for all workers to complete phase 1
    barrier.wait()

In our code, each worker thread performs its phase 1 calculation at its own pace. The barrier ensures that all workers complete phase 1 before any of them move on to phase 2, which needs the combined results from all workers in phase 1.

We're at the end of the quirky stories! Hope you enjoyed it.