Thread-Safe SimpleQueue in Python

You can use a fast and simple thread-safe queue via the queue.SimpleQueue class.

In this tutorial you will discover how to use the thread-safe SimpleQueue class in Python.

Let’s get started.

Need for a Thread-Safe Simple Queue

A thread is a thread of execution in a computer program.

Every Python program has at least one thread of execution called the main thread. Both processes and threads are created and managed by the underlying operating system.

Sometimes we may need to create additional threads in our program in order to execute code concurrently.

Python provides the ability to create and manage new threads via the threading module and the threading.Thread class.

You can learn more about Python threads in the guide:

Threading in Python: The Complete Guide

In concurrent programming, we often need to share data between threads.

One approach to sharing data is to use a queue.

Python provides a number of thread-safe queues in the queue module, such as the queue.SimpleQueue class.

What is the SimpleQueue and how can we use it in Python?

How to Use the SimpleQueue

Python provides a simple thread-safe queue in the queue.SimpleQueue class.

Thread-safe means that it can be used by multiple threads to put and get items concurrently without a race condition.

The “Simple” in the name of the class refers to it providing a minimal interface compared to other queues in the “queue” module.

This minimal interface includes:

Checking the size via qsize().
Check if empty via empty().
Add an item to the queue via put() and put_nowait().
Get an item from the queue via get() and get_nowait().

Unlike the queue.Queue class, the queue.SimpleQueue does not provide the ability to limit the capacity of the queue and as such does not allow threads to block while putting items on the queue. It also does not provide the ability for consumer threads to mark tasks as done and have threads join the queue until all tasks are marked done.

Next, let’s take a closer look at this interface.

SimpleQueue Interface

We can create a new instance of the queue via the constructor which will be unbounded in size by default.

...

# create a new queue

queue = queue.SimpleQueue()

Items are added to the end of the queue and retrieved from the front of the queue in a first-in, first-out or FIFO order.

We can add items to the queue by calling the put() function or the put_nowait() function.

For example:

...

# add an item to the queue (without blocking)

queue.put(item)

Adding items to the queue does not block and does not use a timeout, although the function does take a “block” and “timeout” to match the queue.Queue interface, but these arguments have no effect.

Put item into the queue. The method never blocks and always succeeds (except for potential low-level errors such as failure to allocate memory). The optional args block and timeout are ignored and only provided for compatibility with Queue.put().
— SimpleQueue Objects

Therefore, put_nowait() is equivalent to put().

...

# add an item to the queue without blocking

queue.put_nowait(item)

Items can be retrieved from the queue by calling the get() function.

...

# get an item from the queue

item = queue.get()

By default, the call to get() will block until an item is available to be returned. Therefore, the above is equivalent to the following:

...

# get an item from the queue

item = queue.get(block=True, timeout=None)

A timeout can be used when waiting for an item on the queue via the “timeout” argument. If the timeout expires before an item is available, then an queue.Empty exception is raised.

...

# get an item from the queue with a timeout

try:

item = queue.get(timeout=1)

except Empty:

# no item...

The “block” argument can be set false, which means the function will return immediately if there is an item to retrieve, otherwise it will raise an queue.Empty exception.

...

# get an item from the queue without blocking

try:

item = queue.get(block=False)

except Empty:

# no item...

This is equivalent to calling the get_nowait() function, for example:

...

# get an item from the queue without blocking

try:

item = queue.get_nowait()

except Empty:

# no item...

The number of items in the queue can be queried via the qsize() function.

For example:

...

# check the size of the queue

size = queue.qsize()

We can check if the queue has no values via the empty() function.

For example

...

# check if the queue is empty

if queue.empty():

# ...

Next, let’s take a look at the reentrant capability of the SimpleQueue class.

SimpleQueue is Reentrant

Unlike the queue.Queue class, the get() and put() functions on the queue supports reentry.

Reentrancy in concurrency refers to a capability from concurrent programming where a function call can be interrupted in the middle of execution and another function call before the first function call has finished.

For example, a reentrant lock provided by threading.RLock class can be acquired by a thread, then acquired again by the same thread.

The queue.SimpleQueue is reentrant.

This method has a C implementation which is reentrant. That is, a put() or get() call can be interrupted by another put() call in the same thread without deadlocking or corrupting internal state inside the queue. This makes it appropriate for use in destructors such as __del__ methods or weakref callbacks.
— SimpleQueue Objects

In the queue.SimpleQueue, reentrant refers to the ability to call get() while calling put(), or the reverse, calling put() while calling get().

This can happen in obscure situations.

For example, see:

queue.Queue() is not reentrant, so signals and GC can cause deadlocks

Now that we know how to use the queue.SimpleQueue class, let’s look at some worked examples.

Example of Using the SimpleQueue

We can explore how to use the queue.SimpleQueue class with a worked example.

In this example, we will create a producer thread that will generate ten random numbers and put them on the queue. We will also create a consumer thread that will get numbers from the queue and report their values.

The queue.SimpleQueue provides a simple way to allow these simple producer and consumer threads to communicate data with each other.

First, we can define the function to be executed by the producer thread.

The task will iterate ten times in a loop. Each iteration, it will generate a new random value between 0 and 1 via the random.random() function. It will then block for that fraction of a second, then put the value on the queue.

Once the task is complete it will put the value None on the queue to signal to the consumer that there is no further work.

The producer() function below implements this by taking the queue instance as an argument.

# generate work

def producer(queue):

print('Producer: Running')

# generate work

for i in range(10):

# generate a value

value = random()

# block

sleep(value)

# add to the queue

queue.put(value)

# all done

queue.put(None)

print('Producer: Done')

Next, we can define the function to be executed by the consumer thread.

The task will loop forever. Each iteration, it will get an item from the queue and block if there is no item yet available.

If the item retrieved from the queue is the value None, then the task will break the loop and terminate the thread. Otherwise, the value is reported.

The consumer() function below implements this and takes the queue instance as an argument.

# consume work

def consumer(queue):

print('Consumer: Running')

# consume work

while True:

# get a unit of work

item = queue.get()

# check for stop

if item is None:

break

# report

print(f'>got {item}')

# all done

print('Consumer: Done')

Finally, in the main thread we can create the shared queue instance.

...

# create the shared queue

queue = SimpleQueue()

We can then configure and start the consumer thread, which will patiently wait for work to arrive on the queue.

...

# start the consumer

consumer = Thread(target=consumer, args=(queue,))

consumer.start()

Then we can configure and start the producer thread, which will generate work and add it to the queue for the consumer to retrieve.

...

# start the producer

producer = Thread(target=producer, args=(queue,))

producer.start()

The main thread will then block until both the producer and consumer threads terminate, then terminate itself.

...

# wait for all threads to finish

producer.join()

consumer.join()

Tying this together, the complete example of using the queue.SimpleQueue is listed below.

# SuperFastPython.com

# example of using the simple queue

from time import sleep

from random import random

from threading import Thread

from queue import SimpleQueue

# generate work

def producer(queue):

print('Producer: Running')

# generate work

for i in range(10):

# generate a value

value = random()

# block

sleep(value)

# add to the queue

queue.put(value)

# all done

queue.put(None)

print('Producer: Done')

# consume work

def consumer(queue):

print('Consumer: Running')

# consume work

while True:

# get a unit of work

item = queue.get()

# check for stop

if item is None:

break

# report

print(f'>got {item}')

# all done

print('Consumer: Done')

# create the shared queue

queue = SimpleQueue()

# start the consumer

consumer = Thread(target=consumer, args=(queue,))

consumer.start()

# start the producer

producer = Thread(target=producer, args=(queue,))

producer.start()

# wait for all threads to finish

producer.join()

consumer.join()

Running the example first creates the shared queue instance.

Next, the consumer thread is started and passed the queue instance. Then the producer thread is started and the main thread blocks until the worker threads terminate.

The producer thread generates a new random value each iteration of the task, blocks and adds it to the queue. The consumer thread waits on the queue for items to arrive, then consumes them one at a time, reporting their value.

Finally, the producer task finishes, a None value is put on the queue and the thread terminates. The consumer thread gets the None value, breaks its loop and also terminates.

This highlights how the queue.SimpleQueue can be used to share data easily between a producer and consumer threads.

A sample output of the program is listed below. Note, your specific output will differ given the use of random numbers.

Consumer: Running

Producer: Running

>got 0.49138927575578184

>got 0.9089938160636776

>got 0.9000414387422191

>got 0.69580688995332

>got 0.7816543465569435

>got 0.2853295879765214

>got 0.6840369305347521

>got 0.2025555428328133

>got 0.5096460791785361

Producer: Done

>got 0.3847119182144745

Consumer: Done

Next, let’s look at how we might get values from the queue without blocking.

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

Learn more

Using the SimpleQueue Without Blocking

We can get values from the queue without blocking.

This might be useful if we wish to use busy waiting in the consumer task to check other state or perform other tasks while waiting for data to arrive on the queue.

You can learn more about busy waiting in the tutorial:

How to Use Busy Waiting in Python

We can update the example from the previous section to get items from the queue without blocking.

This can be achieved by setting the “blocking” argument to False when calling get().

For example:

...

# get a value from the queue without blocking

item = queue.get(block=False)

The get() function will return immediately.

If there is a value in the queue to retrieve, then it is returned. Otherwise, if the queue is empty, then a queue.Empty exception is raised, which can be handled.

In this case, if there is no value to get from the queue, we report a message and sleep for a fraction of a second. We will then continue which will jump back to the start of the consumer busy waiting loop.

For example:

...

# get a unit of work

try:

item = queue.get(block=False)

except Empty:

print('Consumer: got nothing, waiting a while...')

sleep(0.5)

continue

The updated version of the consumer() function with this change is listed below.

# consume work

def consumer(queue):

print('Consumer: Running')

# consume work

while True:

# get a unit of work

try:

item = queue.get(block=False)

except Empty:

print('Consumer: got nothing, waiting a while...')

sleep(0.5)

continue

# check for stop

if item is None:

break

# report

print(f'>got {item}')

# all done

print('Consumer: Done')

Tying this together, the complete example is listed below.

# SuperFastPython.com

# example of using the simple queue without blocking

from time import sleep

from random import random

from threading import Thread

from queue import SimpleQueue

from queue import Empty

# generate work

def producer(queue):

print('Producer: Running')

# generate work

for i in range(10):

# generate a value

value = random()

# block

sleep(value)

# add to the queue

queue.put(value)

# all done

queue.put(None)

print('Producer: Done')

# consume work

def consumer(queue):

print('Consumer: Running')

# consume work

while True:

# get a unit of work

try:

item = queue.get(block=False)

except Empty:

print('Consumer: got nothing, waiting a while...')

sleep(0.5)

continue

# check for stop

if item is None:

break

# report

print(f'>got {item}')

# all done

print('Consumer: Done')

# create the shared queue

queue = SimpleQueue()

# start the consumer

consumer = Thread(target=consumer, args=(queue,))

consumer.start()

# start the producer

producer = Thread(target=producer, args=(queue,))

producer.start()

# wait for all threads to finish

producer.join()

consumer.join()

Running the example creates the shared queue, then starts the consumer and producer threads, as before.

The producer thread will generate, block and add items to the queue.

The consumer thread will attempt to get a value from the queue. If there is no value to retrieve, a queue.Empty exception is raised and handled by reporting a message, sleeping for a fraction of a second then starting the busy waiting loop again.

Otherwise, if there is a value in the queue, the consumer will retrieve it and report it as per normal.

A sample output of the program is listed below. Note, your specific output will differ given the use of random numbers.

We can see the messages from the consumer thread busy waiting for new data to arrive in the queue.

This highlights how to get items from the queue.SimpleQueue without blocking.

Consumer: Running