Low-Level Design · lesson 15 of 18
Beginner
4Intermediate
10- 5 Design a Parking Lot: Allocating and Freeing Spots
- 6 Design an Elevator System: Scheduling Requests
- 7 Design a Library Management System: Borrowing and Returning Books
- 8 Design an Expense-Splitting App: Balances and Minimal Settlements
- 9 Design a URL Shortener: Sequential Codes vs Real Hash Collisions
- 10 Wrapping Objects: Adapter, Facade, and Proxy in Python
- 11 The Composite Pattern in Python: Sizing a File Tree Without isinstance
- 12 The State Pattern in Python: Refactoring a Vending Machine Into State Classes
- 13 Undo and Redo in Python: the Memento and Command Patterns Compared
- 14 Chain of Responsibility in Python: Expense Approvals Without an if/elif Ladder
Advanced
4- 15 Concurrency in LLD: A Real Race Condition in the Parking Lot, Then the Fix
- 16 The Observer Pattern in Practice: A Stock Ticker Notifying Multiple Displays
- 17 Design Patterns That Are Just Language Features: Strategy, Command, Iterator
- 18 Abstract Factory and Prototype in Python: Matching Families and Cloning Objects
Also part of Low-Level Design
12- + SOLID Principles in Java from Java
- + Composition vs Inheritance in Java from Java
- + Design Patterns in Java from Java
- + Python Design Patterns from Python
- + Design Patterns in TypeScript from TypeScript
- + JavaScript Design Patterns from JavaScript
- + Design Patterns in C# from C#
- + Design Patterns in C++ from C++
- + Design Patterns in Go from Go
- + Design Patterns in Rust from Rust
- + Abstract Classes in Java from Java
- + Sealed Classes and Interfaces in Java from Java
Concurrency in LLD: A Real Race Condition in the Parking Lot, Then the Fix
Two threads check the same spot, both see it free, and both take it — a real double-booking bug reproduced on purpose, then closed with a threading.Lock.
By SkillByExample · Published
Every LLD example so far has run single-threaded — real systems don’t get that luxury.
The bug: two threads, one spot, both win
import threading
import time
class UnsafeParkingLot:
def __init__(self):
self.occupied = False
def allocate(self, vehicle_id, results):
if not self.occupied:
time.sleep(0.05) # simulates work between checking and acting
self.occupied = True
results.append(f"{vehicle_id} got the spot")
else:
results.append(f"{vehicle_id} found the spot full")
lot = UnsafeParkingLot()
results = []
t1 = threading.Thread(target=lot.allocate, args=("car-1", results))
t2 = threading.Thread(target=lot.allocate, args=("car-2", results))
t1.start(); t2.start(); t1.join(); t2.join()
for r in sorted(results):
print(r)
print(f"Spots given out: {sum('got the spot' in r for r in results)}")
$ python race_condition.py
car-1 got the spot
car-2 got the spot
Spots given out: 2
One spot, two cars, both “got” it. Both threads read self.occupied as False before either
one has a chance to set it to True — the sleep(0.05) guarantees this interleaving happens
every time this runs, but the same bug can occur without any sleep at all, just less
predictably, whenever two threads check-then-act on shared state without coordination.
The fix: a lock around check-and-act together
import threading
import time
class SafeParkingLot:
def __init__(self):
self.occupied = False
self._lock = threading.Lock()
def allocate(self, vehicle_id, results):
with self._lock:
if not self.occupied:
time.sleep(0.05)
self.occupied = True
results.append(f"{vehicle_id} got the spot")
else:
results.append(f"{vehicle_id} found the spot full")
lot = SafeParkingLot()
results = []
t1 = threading.Thread(target=lot.allocate, args=("car-1", results))
t2 = threading.Thread(target=lot.allocate, args=("car-2", results))
t1.start(); t2.start(); t1.join(); t2.join()
for r in sorted(results):
print(r)
print(f"Spots given out: {sum('got the spot' in r for r in results)}")
$ python fixed_lock.py
car-1 got the spot
car-2 found the spot full
Spots given out: 1
Only one thread can be inside the with self._lock: block at once — whichever thread
acquires it first runs the entire check-sleep-write sequence atomically with respect to the
other thread, which has to wait at the with statement until the lock is released.
Practice
1. Confirm the bug gets worse with three threads instead of two.
import threading, time
class UnsafeParkingLot:
def __init__(self):
self.occupied = False
def allocate(self, vehicle_id, results):
if not self.occupied:
time.sleep(0.05)
self.occupied = True
results.append(f"{vehicle_id} got the spot")
else:
results.append(f"{vehicle_id} found the spot full")
lot = UnsafeParkingLot()
results = []
threads = [threading.Thread(target=lot.allocate, args=(f"car-{i}", results)) for i in range(1, 4)]
for t in threads: t.start()
for t in threads: t.join()
for r in sorted(results):
print(r)
print(f"Spots given out: {sum('got the spot' in r for r in results)}")
$ python three_thread_race.py
car-1 got the spot
car-2 got the spot
car-3 got the spot
Spots given out: 3
All three threads pass the check before any of them finishes the write — the bug scales with however many threads race, not just two.
2. Confirm the lock still protects all three threads correctly.
import threading, time
class SafeParkingLot:
def __init__(self):
self.occupied = False
self._lock = threading.Lock()
def allocate(self, vehicle_id, results):
with self._lock:
if not self.occupied:
time.sleep(0.05)
self.occupied = True
results.append(f"{vehicle_id} got the spot")
else:
results.append(f"{vehicle_id} found the spot full")
lot = SafeParkingLot()
results = []
threads = [threading.Thread(target=lot.allocate, args=(f"car-{i}", results)) for i in range(1, 4)]
for t in threads: t.start()
for t in threads: t.join()
for r in sorted(results):
print(r)
print(f"Spots given out: {sum('got the spot' in r for r in results)}")
$ python three_thread_safe.py
car-1 got the spot
car-2 found the spot full
car-3 found the spot full
Spots given out: 1
Exactly one spot given out no matter how many threads compete — the lock serializes them all through the same check-sleep-write sequence, one at a time.
3. Extend the lock-protected lot to a capacity of 2 spots instead of 1.
import threading, time
class SafeParkingLot:
def __init__(self, capacity):
self.capacity = capacity
self.occupied = 0
self._lock = threading.Lock()
def allocate(self, vehicle_id, results):
with self._lock:
if self.occupied < self.capacity:
time.sleep(0.05)
self.occupied += 1
results.append(f"{vehicle_id} got a spot")
else:
results.append(f"{vehicle_id} found the lot full")
lot = SafeParkingLot(capacity=2)
results = []
threads = [threading.Thread(target=lot.allocate, args=(f"car-{i}", results)) for i in range(1, 4)]
for t in threads: t.start()
for t in threads: t.join()
for r in sorted(results):
print(r)
print(f"Spots given out: {sum('got a spot' in r for r in results)}")
$ python capacity_two.py
car-1 got a spot
car-2 got a spot
car-3 found the lot full
Spots given out: 2
A boolean occupied became an integer counter compared against capacity, but the same lock
pattern (guard the check-and-increment together) still holds it at exactly 2, never 3.
Next: the Observer pattern, solving a concrete problem — a stock ticker that has to notify multiple displays without knowing what they are.