Writing Thread-Safe Python Code

# BROKEN — dict is not thread-safe for concurrent reads/writes
import threading

cache = {}   # Shared across threads

def get_or_compute(key: str):
    if key not in cache:          # Thread A reads: not in cache
        value = expensive_compute(key)   # Both threads compute!
        cache[key] = value        # Race condition: both write
    return cache[key]

# Two threads can both see 'key not in cache' and both compute + write

# FIX — threading.Lock for mutual exclusion
import threading

cache = {}
lock = threading.Lock()

def get_or_compute(key: str):
    with lock:
        if key in cache:
            return cache[key]
        value = expensive_compute(key)  # Only one thread computes
        cache[key] = value
        return value

# FIX 2 — Use threading.RLock for reentrant code (same thread can acquire multiple times)
rlock = threading.RLock()

# FIX 3 — Use concurrent.futures or thread-safe collections
from queue import Queue

task_queue = Queue()   # Thread-safe FIFO

# FIX 4 — cachetools for thread-safe LRU cache
from cachetools import TTLCache
from cachetools.keys import hashkey
import threading

cache = TTLCache(maxsize=1000, ttl=300)  # LRU + TTL
cache_lock = threading.Lock()

def get_or_compute(key: str):
    with cache_lock:
        if key not in cache:
            cache[key] = expensive_compute(key)
        return cache[key]