# Concurrency: A Pythonic Approach 

# Parallelism vs. Concurrency

By default, Python code is executed synchronously, meaning that once the function is called, the calling code blocks execution of the main thread until the function finishes. If the function raises an exception, the calling code can wrap the call site with a `try/except` block to handle the errors. While this is convenient and makes it easier to reason about the code we write, there may be times when we want to utilize more of our available resources, or handle more tasks at once. Enter, concurrency and parallelism:

**Concurrency:** The ability of a system to handle multiple tasks by allowing *overlapping* (but not necessarily simultaneous) execution.

**Parallelism:** The simultaneous execution of multiple tasks that utilize multiple processing units. ([Divakaran](https://realpython.com/python-thread-lock/)). This requires a multicore CPU, multiple CPUs, a GPU, or multiple computers in a cluster (Ramalho, 2022, p.697)

While the two concepts are similar in that they involve the execution of multiple tasks, note that parallelism *requires* simultaneous execution. Concurrency is the umbrella term for structuring a program to complete more than one task at once, and parallelism is a specific form, or subset of concurrency. All parallel systems are concurrent, but not all concurrent systems are parallel (Ramalho, p. 695).

### I/O-bound tasks vs. CPU-bound tasks

When discussing different types of concurrent programming, it's important to understand the distinction between [I/O-bound](https://realpython.com/ref/glossary/io-bound-task/) (input/output) and [CPU-bound](https://realpython.com/ref/glossary/cpu-bound-task/) tasks. A program is said to be CPU-bound if the bottleneck is the CPU's processing power, which occurs in problems involving significant CPU computation (often times operations requiring complex math processing). Examples include: audio or image processing, computer vision, and machine or deep learning. Meanwhile, a task is considered I/O-bound when the bottleneck is the communication between the program and the outside world. Common examples include: reading data from a user input or file, making a network request, interacting with external devices, or database operations.

### Processes vs. Threads

Another important distinction is that of a process and a thread. For these I found the most helpful definitions came from *Fluent Python* (Ramalho, 2022):

**Process:** An instance of a computer program in execution, containing it's own memory space and a portion of the CPU time. Processes are isolated entities that contain their own memory spaces. Processes communicate via pipes, sockets, or memory mapped files by converting Python objects into of raw bytes to pass from one process to another. A drawback with processes is the overhead involved in communication, and fact that not all Python objects can be serialized (converted to raw bytes). Processes can spawn subprocesses, or child processes, which are also isolated from each other and the parent. Processes facilitate [preemptive multitasking](https://www.geeksforgeeks.org/operating-systems/difference-between-preemptive-and-cooperative-multitasking/), in which the OS scheduler *preempts* (suspends execution of each process periodically so that other processes can run) (p.698)

**Thread**: A thread is an execution unit (the general term for objects that execute code concurrently, each with independent state and call stack) within a process. A process initializes a single thread (the main thread), and can spawn more threads to operate concurrently by calling operating system APIs. All threads within a process share the same memory space, which contains the active Python objects, thereby facilitating easy data sharing among threads. The tradeoff here is data corruption when one or more thread attempt to update the same object concurrently. Threads also enable preemptive multitasking in conjunction with the OS scheduler (p.698).

### Types of Concurrent vs. Parallel Programming

**Multithreading** refers to a type of concurrent programming in which a process spawns and orchestrates multiple threads to complete a task. Multithreading implements [preemptive multitasking](https://www.geeksforgeeks.org/operating-systems/difference-between-preemptive-and-cooperative-multitasking/), meaning the OS scheduler is responsible for switching contexts (i.e., deciding which thread is running). Each thread can be executing its own set of instructions, but the the Global Interpreter Lock (GIL) ensures that *only one thread* can be running at any given time. Thus, it is *impossible* to utilize multiple processors with threads in Python. The GIL serves as a [mutex](https://en.wikipedia.org/wiki/Lock_%28computer_science%29) (a type of lock that prevents more than one thread from accessing a state) to prevent multiple threads from executing Python [bytecode](https://docs.python.org/3/glossary.html#term-bytecode) at the same time. This is necessary since CPython (the interpreter that executes the bytecode instructions) does not have thread-safe memory management (2).

**Asynchronous programming** is another form of concurrent programming available through the [asyncio](https://docs.python.org/3/library/asyncio.html) library via `async/await` syntax. A key difference between `asyncio` and multithreading is that `asyncio` implements [cooperative multitasking](https://en.wikipedia.org/wiki/Cooperative_multitasking), meaning there is only a single thread and control is passed between [coroutines](https://realpython.com/ref/glossary/coroutine/) (functions defined with the `async def`; can suspend itself and resume later). When an `await` statement is reached, the current coroutine pauses execution, passing control back to the [event loop](https://docs.python.org/3/library/asyncio-eventloop.html) until the current awaited operation returns. While an in-depth discussion of the `asyncio` event loop is beyond the scope of this article, you can think of the event loop as the manager of all the coroutines, coordinating what task to switch to and when to resume a paused coroutine (3).

[Multiprocessing](https://docs.python.org/3/library/multiprocessing.html) enables us to leverage multiple [CPU cores](https://www.hp.com/us-en/shop/tech-takes/cpu-cores-how-many-do-i-need) via parallelization, with each core capable of handling its own tasks and processing information independently. This is achieved via the `multiprocessing` package in Python, and supports spawning multiple processes, circumventing the GIL since it uses subprocesses rather than threads. The key difference is that these subprocesses **do not** share the same memory space, meaning there is significant overhead cost associated with sharing data between processes.

## Race Conditions, & Deadlock

A common pitfall in multithreaded programming is known as a race condition, which occurs when two or more threads attempt to access and modify the same data at the same time and the final result is dependent on the order in which the threads run. For example, say you have two threads (T1 and T2) trying to access your email address in a system. T1 will update your address while T2 is reading the address as it’s been requested. If T1 and T2 are executed simultaneously, T2 could return your old (outdated) email, or the correct (updated) email, depending on which completes first. [Deadlock](https://link.springer.com/rwe/10.1007/978-0-387-09766-4_282) is another concurrent computing phenomenon, which can arise when two or more processes or threads are blocked because each one is waiting for the other to release a resources, bringing progress to a halt.

# Synchronization Primitives

To protect against race conditions when employing concurrent programming, we use [synchronization primitives](https://en.wikipedia.org/wiki/Synchronization_\(computer_science\)#Implementation), which are effectively mechanisms used to control access to shared resources. While these are similar in both multithreading and `asyncio`, the [synchronization primitives for `asyncio`](https://docs.python.org/3/library/asyncio-sync.html) are designed for cooperative multitasking, meaning they are not thread-safe and should only be used within the same event loop. Additionally, `asyncio` primitives do not accept the *timeout* argument, opting for `asyncio.wait_for()` to implement timeouts. The `multiprocessing` module also includes equivalents of all the synchronization primitives from `threading`, but these are not as necessary in a multiprocess program so they will not be discussed in this article. Furthermore, the [official documentation](https://docs.python.org/3/library/multiprocessing.html#programming-guidelines) recommends avoiding shared state and using queues or pipes for communication between processes rather than lower level synchronization primitives.

## Multithreading

### **Locks and RLock**

A lock is used to allow only one thread to access a resource at a time, such that once the lock is acquired, no other threads can acquire the lock until the lock is released. locks are either locked or unlocked, and can be acquired by calling the `Lock.acquire()` method. The `release()` method, is used to unblock execution of other threads when called by the thread holding a locked lock (but will raise a `RuntimeError` if called on an unlocked lock). locks can be used as a context manager to automate acquisition and releasing of locks as shown below:

```python
import threading 
import time
from concurrent.futures import ThreadPoolExecutor

lock = threading.Lock()
class UserAccount:
    def __init__(self):
        self.name = "John Doe"
        self.email = "John@gmail.com"
        self.account_lock = threading.Lock()
 
    def get_email(self):
        with self.account_lock:   
            return self.email
    
    def set_email(self, email):
        with self.account_lock:
            self.email = email

with ThreadPoolExecutor(max_workers=3) as executor:
    executor.submit(set_email, "Johndoe@gmail.com")
    executor.submit(get_email)
    executor.submit(set_email, "John@gmail.com")
```

One drawback with the regular lock is that if the same thread attempts to acquire the lock it already holds, a deadlock will occur. This could happen if within your locked function you invoke another function containing the same lock. Enter, the RLock, or reentrant lock. This synchronization primitive allows the same thread to acquire a lock multiple times before releasing it, preventing deadlock in situations where a thread needs to re-enter a locked resource. The tradeoff is increased overhead since the RLock has to keep track of how many times the same thread has acquired a lock, so it should only be used when necessary (4).

### Semaphores

The [semaphore](https://en.wikipedia.org/wiki/Semaphore_\(programming\)) was invented in the early 1960's by someone every CS student is familiar with: Edsger W. Dijkstra.

A semaphore is a type of atomic counter which guarantees that the OS will not interrupt the thread in the middle of incrementing or decrementing the counter. The internal counter is incremented/decremented with the `release()` and `acquire()` methods, respectively. Semaphores are frequently used to protect a resource with limited capacity, such as a connection pool (4). Semaphores are constructed by passing in the max number of concurrent threads acquiring it. Semaphores can be used as context managers, entering with a successful `acquire()` call and automatically calling `release()` when exiting the with block.

The example below (adapted from [asyncio.Semaphore: Practical Guide with Real-World Use Cases](https://www.soumendrak.com/blog/semaphores-python-async-programming/)) highlights this use case by utilizing a semaphore to cap the number of queries that can execute concurrently using the `max_connections` variable.

```python
import threading
from concurrent.futures import ThreadPoolExecutor
import psycopg2
from psycopg2 import pool


class DatabasePool:
    def __init__(self, dsn, max_connections=5):
        self.dsn = dsn
        self.semaphore = threading.Semaphore(max_connections)
        self.pool = None

    def init_pool(self):
        self.pool = psycopg2.pool.ThreadedConnectionPool(
            minconn=1,
            maxconn=5,
            dsn=self.dsn,
        )

    def query(self, sql, *args):
        with self.semaphore:
            conn = self.pool.getconn()
            try:
                with conn.cursor() as cur:
                    cur.execute(sql, args)
                    return cur.fetchall()
            finally:
                self.pool.putconn(conn)

    def close(self):
        self.pool.closeall()


def main():
    db = DatabasePool("postgresql://user:password@localhost/database")
    db.init_pool()

    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(db.query, "SELECT * FROM users WHERE id = %s", i)
            for i in range(1, 11)
        ]
        results = [f.result() for f in futures]

    db.close()


if __name__ == "__main__":
    main()
```

### Events

Events are objects that allow threads to "communicate" via an internal flag that defaults to `False`, but can be set to `True` by calling the `set()` method, or reset to `False` by calling `clear()`. Individual threads can wait for the flag using the `wait()` method, which blocks execution until the flag is set. Flags can be used to coordinate actions across multiple threads, such as signalling state changes, thereby enabling efficient synchronization management. When a new Costco location opens the often go all out, selling limited-time champagne signed by celebrities and serving caviar samples. The example below uses events to simulate the opening of a new Costco location and the commencement of giving out caviar:

```python
import threading
import time
from concurrent.futures import ThreadPoolExecutor

costco_open = threading.Event()
caviar_open = threading.Event()

def serve_customer(customer_data):
    print(f"{customer_data['name']} is waiting for the Costco to open.")

    costco_open.wait()
    print(f"{customer_data['name']} entered Costco")
    if customer_data["type"] == "VIP_EXPERIENCE":
        print(f"{customer_data['name']} is waiting for caviar to be served.")
        caviar_open.wait()
        print(f"{customer_data['name']} is getting their caviar.")

        # Simulate the time taken for eating caviar
        time.sleep(2)

        print(
            f"{customer_data['name']} finished eating and exited the store"
        )
    else:
        # Simulate the time taken for shopping
        time.sleep(2)
        print(f"{customer_data['name']} has exited the store")

customers = [
    {"name": "Customer 1", "type": "REGULAR_SHOP"},
    {"name": "Customer 2", "type": "VIP_EXPERIENCE"},
    {"name": "Customer 3", "type": "REGULAR_SHOP"},
    {"name": "Customer 4", "type": "VIP_EXPERIENCE"},
]

with ThreadPoolExecutor(max_workers=4) as executor:
    for customer_data in customers:
        executor.submit(serve_customer, customer_data)

    print("Costco manager is preparing to open the store.")
    time.sleep(2)
    print("Costco is now open!")
    costco_open.set()  # Signal that the new location is open

    time.sleep(3)
    print("Caviar is now being served!")
    caviar_open.set()

print("All customers have completed their experiences.")
```

### Conditional Waiting

A condition object is built on top of a Lock or RLock and supports additional functionality allowing threads to wait for certain conditions to be met, and signal other threads of condition changes.

**Methods associated with Condition objects**:

*   `acquire()`: Acquire the underlying lock; must be called before a thread can wait on or signal a condition
    
*   `release()`: Releases the underlying lock
    
*   `wait(timeout=None)`: Blocks the thread until it’s notified, or a specific timeout occurs. The lock is released before blocking and reacquired upon notification or when timeout expires. Useful when a thread needs to wait for a specific condition to be true before proceeding
    
*   `notify(n=1)`: Wakes up one of the threads waiting for the condition (if any are waiting). Will select one randomly if multiple threads are waiting
    
*   `notify_all():` Wakes up all threads waiting for the condition
    

Condition objects are useful for coordinating across threads and managing the flow of execution in a multithreaded environment. The following example from from [Real Python](https://realpython.com/python-thread-lock/#conditions-for-conditional-waiting) demonstrates how a condition can be used to synchronize access to the shared `customer_queue` object and signal the `teller_thread` when a new customer arrives.

```python
import random
import threading
import time
from concurrent.futures import ThreadPoolExecutor

customer_available_condition = threading.Condition()

# Customers waiting to be served by the Teller
customer_queue = []

def now():
    return time.strftime("%H:%M:%S")

def serve_customers():
    while True:
        with customer_available_condition:
            # Wait for a customer to arrive
            while not customer_queue:
                print(f"{now()}: Teller is waiting for a customer.")
                customer_available_condition.wait()

            # Serve the customer
            customer = customer_queue.pop(0)
            print(f"{now()}: Teller is serving {customer}.")

        # Simulate the time taken to serve the customer
        time.sleep(random.randint(1, 5))
        print(f"{now()}: Teller has finished serving {customer}.")

def add_customer_to_queue(name):
    with customer_available_condition:
        print(f"{now()}: {name} has arrived at the bank.")
        customer_queue.append(name)

        customer_available_condition.notify()

customer_names = [
    "Customer 1",
    "Customer 2",
    "Customer 3",
    "Customer 4",
    "Customer 5",
]

with ThreadPoolExecutor(max_workers=6) as executor:
    teller_thread = executor.submit(serve_customers)
    for name in customer_names:
        # Simulate customers arriving at random intervals
        time.sleep(random.randint(1, 3))
        executor.submit(add_customer_to_queue, name)
```

### Barriers

Barriers allow groups of threads to wait for each other before continuing execution. It blocks program execution until a specified number of threads reach the barrier

Barrier takes in one required and two optional arguments:

*   `parties`: the # of threads of the barrier object that the wait() method waits for before proceeding
    
*   `action`: callable that will be executed by one of the threads when released
    
*   `timeout`: timeout value for the wait() method
    

Going back to Costco, let's say that we want to use a barrier to only initiate the Grand Opening (open the flood gates, if you will) once all employees are prepared for the stampede:

```python
import random
import threading
import time
from concurrent.futures import ThreadPoolExecutor

employee_barrier = threading.Barrier(3)

def now():
    return time.strftime("%H:%M:%S")

def prepare_for_work(name):
    print(f"{now()}: {name} is preparing their station.")

    # Simulate the delay to prepare the station
    time.sleep(random.randint(1, 3))
    print(f"{now()}: {name} has finished preparing.")

    # Wait for all employees to finish preparing
    employee_barrier.wait()
    print(f"{now()}: {name} is now ready to serve customers.")

employees = ["Front door person 1", "Front door person 2", "Cashier 1", "Cashier 2", "Caviar sample distributor"]

with ThreadPoolExecutor(max_workers=5) as executor:
    for employee_title in employees:
        executor.submit(prepare_for_work, employee_title)

print(f"{now()}: All employees are ready to serve customers.")
```

Barriers are useful when multiple threads need to be in sync with each other before proceeding, or when you need to coordinate the start of a sequence across multiple threads (4). For example, you could use a barrier when threads are working concurrently to compute a set of results to ensure that all results are in before proceeding to the next stage of computation.

## Asyncio

The `asyncio` synchronization primitives function very similarly to those of threading, with the goal of coordinating access to shared resources between tasks. This section will serve as a high-level overview, so refer to the [official docs](https://docs.python.org/3/library/asyncio-sync.html) for more information!

### **Lock**

Locks are used to guarantee *exclusive* access to a shared resource. The recommended utilization is an `async with` statement

```python
lock = asyncio.Lock()

async with lock:
    # access shared state
```

The `acquire()` and `release()`methods are similar to those to `threading.Lock`:

*   `acquire()`waits until the lock is unlocked, then sets it to locked and returns `True`. Only one coroutine can proceed at a time when more than one are waiting for the lock to be unlocked. Acquiring a lock is *fair*, meaning the coroutines proceed in the order they arrived (FIFO)
    

*   `release()` method resets the lock to *unlocked*, and raises a `RuntimeError` if called on an unlocked lock
    
*   The `locked()` method returns `True` if the lock is locked
    

### **Event**

Events are used to notify multiple `asyncio` tasks that an event has happened. events contain an internal flag that can be set to `True` with the `set()` method and reset to `False` with `clear()` method. The `wait()` method blocks execution until flag is set to `True`. When the `set()` method is invoked, all tasks waiting on the event will be immediately awakened.

### **Condition**

Conditions combines functionality of events and locks. Conditions can be initialized by passing in an existing lock, otherwise a new lock will be created automatically. Conditions allow coordinating exclusive access to a shared resource between tasks.

Condition methods include:

*   The `acquire()` method acquires the underlying lock, waiting until the lock is unlocked.
    

*   `notify(n)` and `notify_all()` wake up *n* tasks or all tasks that are waiting on the condition object, respectively.  The lock must be acquired first, otherwise a `RuntimeError` will be raised.
    
*   The `locked()` method returns `True` if the underlying lock is currently acquired
    
*   `release()`releases underlying lock, resulting in a `RuntimeError` if called on an unlocked lock
    
*   `wait()` causes a task to wait, blocking execution until notified
    
*   `wait_for(predicate)`blocks execution until the specified *predicate* (a callable that is interpreted as a boolean) evaluates to`True`
    

### **Semaphore**

Manages an internal counter, decremented by each `acquire()` call and incremented by each `release()` call. Semaphores are instantiated by passing in a *value* (default is 1), specifying the number of tasks that can access the protected resource at a time. The counter can never go below zero, so if `acquire()` is called at 0, it blocks until another task releases it.

### **Bounded Semaphore**

*   A bounded semaphore is a version of a semaphore that raises a `ValueError` if the `release()` method would cause the internal counter to go above the initial value.
    

### **Barrier**

Similar to threads, barriers block execution at a point in the code until a specified number of *parties* reach it, at which point all tasks are unblocked simultaneously.

## Summary: When should we use what?

Key point: neither multithreading nor asynchronous programming will ever increase the power in a system due to the cost of task switching and synchronization. As such, good candidates for these techniques are tasks that involve significant downtime waiting for *external* events (i.e., I/O-bound tasks).

Multithreading is desirable in Python due to preemptive task switching, meaning no additional code is required to context switch. Further, all threads share a state, so no additional overhead is required for threads to access a shared resources such as variables and data structures. However, this is a double-edged sword since multiple threads accessing and shared state can result in [race conditions](https://en.wikipedia.org/wiki/Race_condition) affecting critical sections, and deadlock can arise as a result of protecting such sections. Locks are problematic because it becomes very difficult to reason about your code as more are added (1). Furthermore, locks don't actually "lock" anything– they're just a signal that a thread can check– but if a thread doesn't check to acquire a lock, they can *still access the resource*.

Between async and multithreading, `asyncio` may be desirable to multithreading since you no longer have to deal with locks or worry about arbitrary interruptions. Additionally, the cost of switching contexts is very low because it uses generators under the hood to store the state (1). The disadvantage of the asynchronous approach is that you have to explicitly add keywords such as `yield` or `await` to pass control back to the scheduler and manage the flow of execution. Furthermore, every single thing you do has to be non-blocking, and it requires all involved libraries to be async-compatible (1).

For CPU-bound tasks, we opt for multiprocessing, since such tasks block the event loop when executed asynchronously, and the Python GIL causes threads to perform *worse* than sequential code during intense computations. Indeed, Python is very popular in the modern data science community, supporting compute-intensive tools such as [Project Jupyter](https://jupyter.org/), [TensorFlow](https://www.tensorflow.org/), [PyTorch](https://pytorch.org/), and [Dask](https://www.dask.org/).

**Sources:**

1.  [**Raymond Hettinger, Keynote on Concurrency, PyBay 2017**](https://www.youtube.com/watch?v=9zinZmE3Ogk)
    
2.  [The Python GIL (Global Interpreter Lock)](https://python.land/python-concurrency/the-python-gil)
    
3.  [Python's asyncio: A Hands-On Walkthrough](https://realpython.com/async-io-python/?gad_source=1&gad_campaignid=23282418443&gbraid=0AAAAA_bFrtJLh04S5MREWA282BxKywlUr&gclid=CjwKCAjwnZfPBhAGEiwAzg-VzI08vi1_Hf2VEfAe1vP-vwkJrVhjYlBr4WHg6H_v1bf0pxNut-xqTRoCs74QAvD_BwE)
    
4.  [Python Thread Safety: Using a Lock and Other Techniques](https://realpython.com/python-thread-lock/)
    
5.  Ramalho, L. (2022). *Fluent Python : clear, concise, and effective programming*. O’reilly Media, Inc.
    
    ‌