Mastering Thread Safety in Python: Locks, Conditions, and More

Thread Safety

Thread safety is a concept in multithreaded or multiprocess programming where code that accesses shared data is protected by synchronization mechanisms so that each thread can execute correctly without data corruption.

The main problem arises from thread switching. Imagine a room (process) with 10 candies (resources) and three people (one main thread and two child threads). If person A eats three candies and is pre‑empted, it believes seven remain; later person B eats three more, so when A resumes it still thinks seven are left, while only four actually remain. This unsynchronised access exemplifies a thread‑safety issue.

Consider the following example: a variable num is initialised to 0 and two threads are started—one increments num ten million times, the other decrements it ten million times. The final result is often far from the expected 0 because of race conditions.

import threading

num = 0

def add():
    global num
    for i in range(10_000_000):
        num += 1

def sub():
    global num
    for i in range(10_000_000):
        num -= 1

if __name__ == "__main__":
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)
    subThread01.start()
    subThread02.start()
    subThread01.join()
    subThread02.join()
    print("num result : %s" % num)
    # Sample outputs:
    # num result : 669214
    # num result : -1849179
    # num result : -525674

To solve this problem you must use locks to control the timing of thread switches.

Note that Python’s built‑in container types list, tuple, and dict are themselves thread‑safe for individual operations, so concurrent modifications of these containers do not require additional locking.

Purpose of Locks

Locks are a mechanism provided by Python to let developers control thread switching. By using a lock, thread switches become ordered, making data access and modification predictable and safe.

The threading module offers five common lock types, classified by functionality:

Sync lock: Lock (only one thread can hold it at a time)

Recursive lock: RLock (only one thread can hold it, but the same thread may acquire it multiple times)

Condition lock: Condition (can release any number of waiting threads)

Event lock: Event (releases all waiting threads at once)

Semaphore lock: Semaphore (releases a specific number of waiting threads)

1. Lock() – Sync Lock

Basic introduction

The sync lock is also called a mutual‑exclusion lock. Mutual exclusion means that at any moment only one thread can access the protected resource, but it does not enforce any order of access. Synchronisation builds on mutual exclusion by adding ordering.

Usage example:

import threading

num = 0

def add():
    lock.acquire()
    global num
    for i in range(10_000_000):
        num += 1
    lock.release()

def sub():
    lock.acquire()
    global num
    for i in range(10_000_000):
        num -= 1
    lock.release()

if __name__ == "__main__":
    lock = threading.Lock()
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)
    subThread01.start()
    subThread02.start()
    subThread01.join()
    subThread02.join()
    print("num result : %s" % num)
    # Expected output three times: 0

Using a sync lock makes the code effectively serial, which for CPU‑bound work may be slower than a single‑threaded implementation; the example is only for illustration.

Deadlock

For a sync lock, each acquire() must be matched by a corresponding release(). Repeating acquire() without matching releases leads to deadlock, causing the program to block indefinitely.

import threading

num = 0

def add():
    lock.acquire()  # lock
    lock.acquire()  # deadlock
    global num
    for i in range(10_000_000):
        num += 1
    lock.release()
    lock.release()

def sub():
    lock.acquire()
    lock.acquire()
    global num
    for i in range(10_000_000):
        num -= 1
    lock.release()
    lock.release()

if __name__ == "__main__":
    lock = threading.Lock()
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)
    subThread01.start()
    subThread02.start()
    subThread01.join()
    subThread02.join()
    print("num result : %s" % num)

With statement

Because Lock implements the context‑manager protocol, it can be used with a with statement, which automatically acquires and releases the lock.

import threading

num = 0

def add():
    with lock:
        global num
        for i in range(10_000_000):
            num += 1

def sub():
    with lock:
        global num
        for i in range(10_000_000):
            num -= 1

if __name__ == "__main__":
    lock = threading.Lock()
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)
    subThread01.start()
    subThread02.start()
    subThread01.join()
    subThread02.join()
    print("num result : %s" % num)

2. RLock() – Recursive Lock

Basic introduction

A recursive lock is an upgraded version of the sync lock that allows the same thread to acquire it multiple times, provided that each acquire is matched by a release. Mismatched acquire/release counts still cause deadlock.

Usage example:

import threading

num = 0

def add():
    lock.acquire()
    lock.acquire()
    global num
    for i in range(10_000_000):
        num += 1
    lock.release()
    lock.release()

def sub():
    lock.acquire()
    lock.acquire()
    global num
    for i in range(10_000_000):
        num -= 1
    lock.release()
    lock.release()

if __name__ == "__main__":
    lock = threading.RLock()
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)
    subThread01.start()
    subThread02.start()
    subThread01.join()
    subThread02.join()
    print("num result : %s" % num)

With statement

import threading

num = 0

def add():
    with lock:
        global num
        for i in range(10_000_000):
            num += 1

def sub():
    with lock:
        global num
        for i in range(10_000_000):
            num -= 1

if __name__ == "__main__":
    lock = threading.RLock()
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)
    subThread01.start()
    subThread02.start()
    subThread01.join()
    subThread02.join()
    print("num result : %s" % num)

3. Condition() – Condition Lock

Basic introduction

A condition lock builds on a recursive lock and adds the ability to pause thread execution. Threads can call wait() to block and notify() (or notify_all()) to wake one or more waiting threads.

Note: the number of threads released by a condition lock can be freely chosen.

Example that starts ten sub‑threads, puts them all into a waiting state, and then notifies a user‑specified number of threads to continue:

import threading

currentRunThreadNumber = 0
maxSubThreadNumber = 10

def task():
    global currentRunThreadNumber
    thName = threading.current_thread().name
    condLock.acquire()
    print("start and wait run thread : %s" % thName)
    condLock.wait()
    currentRunThreadNumber += 1
    print("carry on run thread : %s" % thName)
    condLock.release()

if __name__ == "__main__":
    condLock = threading.Condition()
    for i in range(maxSubThreadNumber):
        subThreadIns = threading.Thread(target=task)
        subThreadIns.start()
    while currentRunThreadNumber < maxSubThreadNumber:
        notifyNumber = int(input("Please enter the number of threads that need to be notified to run："))
        condLock.acquire()
        condLock.notify(notifyNumber)
        condLock.release()
    print("main thread run end")

With statement

import threading

currentRunThreadNumber = 0
maxSubThreadNumber = 10

def task():
    global currentRunThreadNumber
    thName = threading.current_thread().name
    with condLock:
        print("start and wait run thread : %s" % thName)
        condLock.wait()
        currentRunThreadNumber += 1
        print("carry on run thread : %s" % thName)

if __name__ == "__main__":
    condLock = threading.Condition()
    for i in range(maxSubThreadNumber):
        subThreadIns = threading.Thread(target=task)
        subThreadIns.start()
    while currentRunThreadNumber < maxSubThreadNumber:
        notifyNumber = int(input("Please enter the number of threads that need to be notified to run："))
        with condLock:
            condLock.notify(notifyNumber)
    print("main thread run end")

4. Event() – Event Lock

Basic introduction

An event lock is based on a condition lock but can only release all waiting threads at once, similar to a traffic light where red stops all cars and green lets them all go.

Example simulating a traffic light:

# Create an event object
eve = threading.Event()
# Red light (clear)
eve.clear()
# Wait for green
# ...
# Green light (set)
eve.set()

import time, threading

def light(eve):
    print(f'Current time:{time.ctime()}, red light ends in 5s!')
    time.sleep(5)
    print(f'Current time:{time.ctime()}, green light on!')
    eve.set()

def car(eve, name):
    print(f'Current time:{time.ctime()}, car {name} waiting at red light')
    eve.wait()
    print(f'Current time:{time.ctime()}, car {name} proceeds')

if __name__ == "__main__":
    eve = threading.Event()
    t1 = threading.Thread(target=light, args=(eve,))
    t1.start()
    for each in 'ABCDE':
        t2 = threading.Thread(target=car, args=(eve, each))
        t2.start()

5. Semaphore() – Semaphore Lock

Basic introduction

A semaphore limits the number of threads that can hold the lock simultaneously. It is useful for modelling a resource with limited capacity.

Example that allows only two threads to run concurrently:

import threading, time

maxSubThreadNumber = 6

def task():
    thName = threading.current_thread().name
    with semaLock:
        print("run sub thread %s" % thName)
        time.sleep(3)

if __name__ == "__main__":
    semaLock = threading.Semaphore(2)  # only two threads at a time
    for i in range(maxSubThreadNumber):
        subThreadIns = threading.Thread(target=task)
        subThreadIns.start()

Lock Relationship Overview

All five lock types are built on the basic sync lock. The recursive lock maintains an internal counter; when the counter is non‑zero the thread cannot be pre‑empted. The condition lock internally uses a low‑level lock (a sync lock) and a high‑level lock (a recursive lock). When wait() is called, the low‑level lock is temporarily released while the high‑level lock remains held until another thread calls notify(), at which point the low‑level lock is reacquired.

Basic Exercises

Condition Lock Application

Goal: create an empty list and have two threads alternately add even and odd numbers so that the final list contains the ordered sequence 1‑100.

import threading

lst = []

def even():
    """Add even numbers"""
    with condLock:
        for i in range(2, 101, 2):
            if len(lst) % 2 != 0:
                lst.append(i)
                condLock.notify()
                condLock.wait()
            else:
                condLock.wait()
                lst.append(i)
                condLock.notify()
        condLock.notify()

def odd():
    """Add odd numbers"""
    with condLock:
        for i in range(1, 101, 2):
            if len(lst) % 2 == 0:
                lst.append(i)
                condLock.notify()
                condLock.wait()
        condLock.notify()

if __name__ == "__main__":
    condLock = threading.Condition()
    addEvenTask = threading.Thread(target=even)
    addOddTask = threading.Thread(target=odd)
    addEvenTask.start()
    addOddTask.start()
    addEvenTask.join()
    addOddTask.join()
    print(lst)

Event Lock Application

Goal: two threads representing poets Li Bai and Du Fu exchange lines of dialogue in turn.

import threading

def libai():
    event.wait()
    print("Li Bai: Old Du, I can't drink any more!")
    event.set()
    event.clear()
    event.wait()
    print("Li Bai: Huh… I fell asleep.")

def dufu():
    print("Du Fu: Old Li, let's have a drink!")
    event.set()
    event.clear()
    event.wait()
    print("Du Fu: Old Li, another pot?")
    print("Du Fu: …Old Li?")
    event.set()

if __name__ == "__main__":
    event = threading.Event()
    t1 = threading.Thread(target=libai)
    t2 = threading.Thread(target=dufu)
    t1.start()
    t2.start()
    t1.join()
    t2.join()