Semaphores (limit concurrency)

What is a semaphore?

A semaphore controls access to a limited resource.

Example: only 3 workers may access an API at once.

Example

semaphore.py

import threading
import time
 
sem = threading.Semaphore(3)
 
 
def worker(i: int) -> None:
    with sem:
        print("start", i)
        time.sleep(0.5)
        print("end", i)
 
 
threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

semaphore.py

import threading
import time
 
sem = threading.Semaphore(3)
 
 
def worker(i: int) -> None:
    with sem:
        print("start", i)
        time.sleep(0.5)
        print("end", i)
 
 
threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

BoundedSemaphore

BoundedSemaphoreBoundedSemaphore raises an error if you release too many times.

Use it to catch bugs.

bounded_semaphore.py

import threading
 
sem = threading.BoundedSemaphore(1)
sem.acquire()
sem.release()
# sem.release()  # would raise ValueError

bounded_semaphore.py

import threading
 
sem = threading.BoundedSemaphore(1)
sem.acquire()
sem.release()
# sem.release()  # would raise ValueError

Semaphores (limit concurrency)

What is a semaphore?

Example

BoundedSemaphore

🧪 Try It Yourself

Exercise 1 – Create a Semaphore

Exercise 2 – Semaphore as Context Manager

Exercise 3 – BoundedSemaphore

Was this page helpful?