Imagine a parking garage with only a few spaces. A semaphore acts like the garage’s available-space counter: each task takes a permit before using a shared resource and returns it afterward. Several tasks may hold permits at once, unlike with a mutex. section 8.5
Internally, a semaphore stores a counter. Acquiring it decreases the counter when it is positive; releasing it increases the counter. When the counter reaches zero, further acquirers block until another task releases a permit. A mutex is a specialized binary semaphore whose counter is limited to 0 or 1. section 8.5
A three-space parking garage
A garage can initialize Semaphore(3), allowing three cars to park concurrently. Ten threads may attempt to enter, but only three obtain permits initially; the rest wait. Each departing car releases its permit, allowing another waiting car to enter. A separate lock protects the shared list of parked cars. section 8.5 section 8.7
# Chapter 8/semaphore.py
import typing as T
import time
import random
from threading import Thread, Semaphore, Lock
TOTAL_SPOTS = 3
class Garage:
def __init__(self) -> None:
self.semaphore = Semaphore(TOTAL_SPOTS) #1
self.cars_lock = Lock() #2
self.parked_cars: T.List[str] = []
def count_parked_cars(self) -> int:
return len(self.parked_cars)
def enter(self, car_name: str) -> None:
self.semaphore.acquire() #1
self.cars_lock.acquire() #2
self.parked_cars.append(car_name)
print(f"{car_name} parked")
self.cars_lock.release() #2
def exit(self, car_name: str) -> None:
self.cars_lock.acquire() #2
self.parked_cars.remove(car_name)
print(f"{car_name} leaving")
self.semaphore.release() #3
self.cars_lock.release() #2
Inspect the Python Garage implementation, which combines a capacity semaphore with a lock protecting the parked-car list. section 8.5
# Chapter 8/semaphore.py
def park_car(garage: Garage, car_name: str) -> None:
garage.enter(car_name) #1
time.sleep(random.uniform(1, 2)) #1
garage.exit(car_name) #1
def test_garage(garage: Garage, number_of_cars: int = 10) -> None:
threads = []
for car_num in range(number_of_cars): #2
t = Thread(target=park_car, #2
args=(garage, f"Car #{car_num}")) #2
threads.append(t) #2
t.start() #2
for thread in threads:
thread.join()
if __name__ == "__main__":
number_of_cars = 10
garage = Garage()
test_garage(garage, number_of_cars) #3
print("Number of parked cars after a busy day:")
print(f"Actual: {garage.count_parked_cars()}\nExpected: 0")
Follow multiple threads as they park, wait, leave, and release permits during a simulated busy day. section 8.5
Use a semaphore when:
- A shared resource has a fixed capacity and several tasks may use available capacity concurrently. section 8.5
- A resource must be acquired before use and released when work finishes. section 8.5
- Avoiding synchronization is possible; synchronization adds cost, so design without it when practical. section 8.7
