Last Updated on September 12, 2022
You can use a semaphore as a latch, as a mutex and as a thread-safe counter.
In this tutorial you will discover advanced usages of the semaphore in Python.
Let’s get started.
Semaphore in Python
A semaphore is a concurrency primitive.
This is one of the oldest synchronization primitives in the history of computer science, invented by the early Dutch computer scientist Edsger W. Dijkstra
— Semaphore Objects
A semaphore is an object that manages an internal counter in a thread-safe manner.
It can be used to limit access to a resource. For example, it can be used to limit the number of concurrent threads performing an action or executing some code, such as:
- Limit concurrent socket connections to avoid overwhelming a remote server.
- Limit concurrent open files to avoid using too much system memory.
Python provides a semaphore via the threading.Semaphore class.
A semaphore can be created and the initial value of the counter specified as an argument.
For example, we may initialize the counter with the number of concurrent accesses to a resource, such as 10:
|
1 2 3 |
... # create a semaphore with an initial value of ten semaphore = threading.Semaphore(10) |
Typically usage involves acquiring the semaphore, performing an action and then releasing the semaphore.
The internal counter within the semaphore is decremented by calls to the acquire() function and incremented by calls to the release() function.
For example:
|
1 2 3 4 5 6 7 |
... # acquire the semaphore (decrement counter) semaphore.acquire() # perform limited action # ... # release the semaphore (increment counter) semaphore.release() |
The same effect can be achieved using the context manager interface that will acquire the semaphore prior to the block and release it automatically when the block is exited.
For example:
|
1 2 3 4 5 6 |
... # acquire the semaphore (decrement counter) with semaphore: # perform limited action # ... # release the semaphore automatically (increment counter) |
In this way, the counter is decremented each time the limited action is performed.
If the counter is zero, calls to acquire() block until the counter is incremented by another thread. This prevents the limited action being performed more than the specified number of times, provided as an argument to the threading.Semaphore class.
You can learn more about the semaphore and its normal usage in this tutorial:
Now that we have seen the normal usage of a semaphore, let’s look at some advanced usages.
Run loops using all CPUs, download your FREE book to learn how.
Advanced Usages of a Semaphore
The semaphore can be used in some alternate ways that may not be obvious.
These include:
- The semaphore can be used as a mutex
- The semaphore can be used as a latch
- The semaphore can be used as a counter.
Let’s take a look at each of these examples in more detail.
Example of Semaphore as a Mutex
A mutual exclusion lock or mutex for short is a concurrency primitive.
It can be used to protect a critical section of code from race conditions.
For example, a critical section can be protected by parallel execution by requiring that a thread first acquire the lock before executing the section. Once the section is executed, the lock can then be released.
While the lock is held, no other thread can acquire the lock and therefore cannot execute the critical section. Instead, other threads must wait until the lock becomes available after which they can attempt to acquire it.
Python provides the threading.Lock class that implements a mutex lock.
For example:
|
1 2 3 4 5 6 7 8 |
... # create a lock lock = threading.Lock() # ... # protect a critical section of code with lock: # execute critical section # ... |
You can learn more about mutex locks in this tutorial:
The semaphore can be used as a mutex lock.
This can be achieved by creating a new semaphore with an initial value of one.
For example:
|
1 2 3 |
... # create the semaphore semaphore = threading.Semaphore(1) |
This is called a binary semaphore, as it can either have a count of one or zero.
- A Value of One: The semaphore can be acquired.
- A Value of Zero: The semaphore cannot be acquired, and callers must block.
The critical section of code can then be protected by the semaphore. Each time the critical section is executed, a thread must first acquire the semaphore.
For example:
|
1 2 3 4 5 |
... # acquire the semaphore with semaphore: # execute the critical section # ... |
This will decrement the counter from 1 to 0. Any other thread attempting to acquire the semaphore will block until the semaphore is released.
We can demonstrate using a semaphore as a mutex lock with a worked example.
First, we can define a task that has a critical section that must be protected. The task will simply report a message and then block for a random fraction of a second.
The critical section will be protected by the semaphore acting as a mutex, ensuring that only one thread can execute the task at any time.
The task() function below implements this, taking the semaphore and an integer to identify the thread uniquely as arguments.
|
1 2 3 4 5 6 7 |
# task for new thread def task(semaphore, identifier): # acquire the semaphore with semaphore: # complete task print(f'Thread {identifier} working') sleep(random()) |
In the main thread, we can create the semaphore and configure it as a mutex lock with a count of one.
|
1 2 3 |
... # create shared semaphore semaphore = Semaphore(1) |
Next, we can create ten threads to execute our task concurrently, each provided the shared semaphore and a unique id from 0 to 9.
|
1 2 3 |
... # configure threads threads = [Thread(target=task, args=(semaphore,i)) for i in range(10)] |
Next, the main thread will start all ten threads to run at the same time and execute the task with the critical section.
|
1 2 3 4 |
... # start threads for thread in threads: thread.start() |
The main thread will then block until all new task threads have terminated.
|
1 2 3 4 |
... # wait for threads to terminate for thread in threads: thread.join() |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# SuperFastPython.com # example of a semaphore acting as a mutex from random import random from time import sleep from threading import Thread from threading import Semaphore # task for new thread def task(semaphore, identifier): # acquire the semaphore with semaphore: # complete task print(f'Thread {identifier} working') sleep(random()) # create shared semaphore semaphore = Semaphore(1) # configure threads threads = [Thread(target=task, args=(semaphore,i)) for i in range(10)] # start threads for thread in threads: thread.start() # wait for threads to terminate for thread in threads: thread.join() |
Running the example first creates the shared semaphore.
Next, the ten worker threads are created, all configured to execute the task. The threads are started and then the main thread blocks until the new threads terminate.
Each thread executes the custom task function, first attempting to acquire the semaphore.
Only a single thread is able to acquire the semaphore at a time, so all other threads are blocked.
Threads execute the critical section one at a time, block for a random interval of a second, then release the semaphore.
A sample of the output is provided below. Your results may differ given the use of random numbers.
|
1 2 3 4 5 6 7 8 9 10 |
Thread 0 working Thread 1 working Thread 2 working Thread 3 working Thread 4 working Thread 5 working Thread 6 working Thread 7 working Thread 8 working Thread 9 working |
Next, let’s look at how we might use the semaphore as a latch.
Free Python Threading Course
Download your FREE threading PDF cheat sheet and get BONUS access to my free 7-day crash course on the threading API.
Discover how to use the Python threading module including how to create and start new threads and how to use a mutex locks and semaphores
Example of Semaphore as a Binary Latch
A semaphore can be used as a latch.
A latch is a concurrency primitive where threads decrement a shared counter until a value of zero is reached, at which point other threads waiting on the latch are notified.
This type of latch is often called a count down latch or simply latch for short.
It is like a barrier in that it involves threads decrementing a shared counter and notifying threads when the counter reaches zero. In a barrier, the threads that decrement the counter must also wait for the counter to reach zero, whereas in a latch, the threads that decrement the counter are not blocked and are separate from the thread or threads waiting for the counter to reach zero.
You can learn more about how to develop a countdown latch from scratch in this tutorial:
We can use a semaphore as a simple latch called a binary latch. This is a latch that is initially closed but may be opened by one or more threads.
This can be achieved by creating a semaphore in the blocked position, with an initial value of zero.
For example:
|
1 2 3 |
... # initialize the latch latch = threading.Semaphore(0) |
The latch will open when any thread increments the counter in the semaphore by calling the release() function.
For example:
|
1 2 3 |
... # release the latch (increment counter) latch.release() |
This type of binary latch is helpful when you want to block a thread until a specific condition is met, but are not concerned about which thread meets the condition first.
It may also be useful between just two threads, one that is blocked by the latch and another that will release the latch.
Note, we are using the semaphore as a simple monitor or condition, e.g. the wait and notify pattern. You can learn more about condition objects in this tutorial:
We can demonstrate this with a worked example.
First, we can define a task that will perform some action, then trigger the latch, then carry on with some other work.
The task() function below implements this, taking the shared latch as an argument.
|
1 2 3 4 5 6 7 8 9 10 |
# task for new thread def task(latch): # complete task sleep(2) # indicate complete latch.release() # do some other work sleep(1) # report all done print('Task all done') |
Next, we can create the shared latch in the main thread.
|
1 2 3 |
... # create shared latch latch = Semaphore(0) |
We can then create a new thread configured to execute our task function and start executing it.
|
1 2 3 4 |
... # start a new thread to complete some work thread = Thread(target=task, args=(latch,)) thread.start() |
The main thread will then block, waiting for the task function to complete a unit of work and trigger the latch to open.
|
1 2 3 4 5 6 |
... # wait for first part of the task to be done print('Main waiting on task') latch.acquire() # continue on print('Main continuing on...') |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
# SuperFastPython.com # example of a semaphore acting as a latch from time import sleep from threading import Thread from threading import Semaphore # task for new thread def task(latch): # complete task sleep(2) # indicate complete latch.release() # do some other work sleep(1) # report all done print('Task all done') # create shared latch latch = Semaphore(0) # start a new thread to complete some work thread = Thread(target=task, args=(latch,)) thread.start() # wait for first part of the task to be done print('Main waiting on task') latch.acquire() # continue on print('Main continuing on...') |
Running the example first creates the shared semaphore operating as a latch.
Next, the new thread is configured and started.
The main thread then blocks on the latch, waiting for it to open.
The new thread blocks for a moment, then releases the latch. It then carries on with other work it needs to complete with a blocking call and a print statement.
The main thread is notified when the latch is opened and is able to carry on executing.
|
1 2 3 |
Main waiting on task Main continuing on... Task all done |
Next, let’s look at how we may use a semaphore as a thread-safe counter.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of a Semaphore as a Counter
A semaphore may be used as a thread-safe counter.
Internally, the semaphore protects a counter with a lock and a condition object. As such, we can use the semaphore as a shared counter.
For example, we can create the semaphore with an initial count of zero.
|
1 2 3 |
... # create shared semaphore semaphore = Semaphore(0) |
We can then increment it in a thread-safe manner by calling the release() function.
For example:
|
1 2 3 |
... # increment counter semaphore.release() |
We might also want to increase the count by a specific amount, for example:
|
1 2 3 |
... # add to the counter semaphore.release(10) |
The value of the counter can be reported via the _value private attribute on the threading.Semaphore class.
For example:
|
1 2 3 |
... # report counter value print(semaphore._value) |
The counter can be decremented by calling acquire(), but if the counter has a value of zero (or less), it will block. Therefore care must be taken when decrementing.
For example:
|
1 2 3 |
... # decrement counter semaphore.acquire() |
We cannot add a negative value to the counter by calling release() with a value less than one. This will result in a ValueError.
For example:
|
1 2 3 |
... # decrement the counter semaphore.release(-10) # raises a ValueError |
A downside of using a semaphore as a thread-safe counter is that each call to release() will attempt to notify any waiting threads. This adds some minor undesirable computational overhead.
It would be better to develop a thread-safe counter using a custom class.
You can learn more about developing a thread-safe counter from scratch in this tutorial:
We can demonstrate using a semaphore as a thread-safe counter.
First, we can define a task function that will iterate a number of times, block for a random fraction of a second, then increment the counter and report its value.
The task() function below implements this, taking the shared counter as an argument.
|
1 2 3 4 5 6 7 8 9 |
# task for incrementing the counter def task(counter): for i in range(10): # block sleep(random()) # increment counter counter.release() # report counter value print(counter._value) |
The main thread can then create the shared counter with an initial value of zero.
|
1 2 3 |
... # create shared semaphore counter = Semaphore(0) |
We can then create 1,000 new threads configured to execute our task() function.
|
1 2 3 |
... # configure increment threads threads = [Thread(target=task, args=(counter,)) for _ in range(1000)] |
These new threads can then be started, and the main thread can block until the threads terminate.
|
1 2 3 4 5 6 7 |
... # start threads for thread in threads: thread.start() # wait for threads to terminate for thread in threads: thread.join() |
Finally, the value of the counter can be reported.
|
1 2 3 |
... # report counter value print(f'Counter Value: {counter._value}') |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
# SuperFastPython.com # example of a semaphore acting as a thread-safe counter from random import random from time import sleep from threading import Thread from threading import Semaphore # task for incrementing the counter def task(counter): for i in range(10): # block sleep(random()) # increment counter counter.release() # report counter value print(counter._value) # create shared semaphore counter = Semaphore(0) # configure increment threads threads = [Thread(target=task, args=(counter,)) for _ in range(1000)] # start threads for thread in threads: thread.start() # wait for threads to terminate for thread in threads: thread.join() # report counter value print(f'Counter Value: {counter._value}') |
Running the example first creates the shared thread-safe counter.
Next, the 1,000 threads are created and started, then the main thread blocks until they are finished.
Each thread increments the shared counter by 10 in a thread-safe manner and reports the current value of the counter each iteration.
The new threads terminate and the final value of the counter is reported.
In this case it has a value of 10,000, which is the correct value given 1,000 threads each incrementing by 10, e.g. 1,000 * 10.
|
1 2 3 4 5 6 7 8 9 10 11 12 |
... 9991 9992 9993 9994 9995 9996 9997 9998 9999 10000 Counter Value: 10000 |
Further Reading
This section provides additional resources that you may find helpful.
Python Threading Books
- Python Threading Jump-Start, Jason Brownlee (my book!)
- Threading API Interview Questions
- Threading Module API Cheat Sheet
I also recommend specific chapters in the following books:
- Python Cookbook, David Beazley and Brian Jones, 2013.
- See: Chapter 12: Concurrency
- Effective Python, Brett Slatkin, 2019.
- See: Chapter 7: Concurrency and Parallelism
- Python in a Nutshell, Alex Martelli, et al., 2017.
- See: Chapter: 14: Threads and Processes
Guides
- Python Threading: The Complete Guide
- Python ThreadPoolExecutor: The Complete Guide
- Python ThreadPool: The Complete Guide
APIs
References
Takeaways
You now know some advanced usages of a semaphore in Python.
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by vikram sundaramoorthy on Unsplash
Leave a Reply